Agentic Planning Strategies — The $12k Mistake We Made with ReAct Loops on a Fraud Pipeline

import os
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.callbacks import OpenAICallbackHandler
from langchain.schema import SystemMessage, HumanMessage

# Production setup: always set a cost budget
llm = ChatOpenAI(model="gpt-4o", temperature=0, max_tokens=500)

@tool
def check_blacklist(ip: str) -> str:
    """Check if an IP is in the blacklist."""
    # Simulated API call
    return "not found"

@tool
def get_transaction_history(user_id: str) -> str:
    """Get recent transactions for a user."""
    return "txn_123: $50, txn_456: $2000"

# Build the agent
prompt = SystemMessage(content="You are a fraud investigator. Use tools to gather evidence. Be concise.")
agent = create_react_agent(llm, [check_blacklist, get_transaction_history], prompt)

# The fix: explicit max_iterations and a callback for cost tracking
cb = OpenAICallbackHandler()
agent_executor = AgentExecutor(
    agent=agent,
    tools=[check_blacklist, get_transaction_history],
    max_iterations=10,  # Hard cap — never set to None
    early_stopping_method="force",
    callbacks=[cb],
    verbose=True
)

# Run with a timeout guard (use asyncio.wait_for in async code)
import signal
class TimeoutError(Exception):
    pass

def handler(signum, frame):
    raise TimeoutError("Agent took too long")

signal.signal(signal.SIGALRM, handler)
signal.alarm(30)  # 30 second timeout

try:
    result = agent_executor.invoke({"input": "Investigate user_id=abc123 for fraud"})
    print(f"Result: {result}")
    print(f"Total tokens used: {cb.total_tokens}")
    if cb.total_tokens > 2000:
        print("WARNING: token budget exceeded, consider reducing max_iterations")
except TimeoutError:
    print("Agent timed out — check for infinite loop")
finally:
    signal.alarm(0)

import json
from datetime import datetime, timedelta
from langchain_openai import ChatOpenAI
from langchain.tools import tool

llm = ChatOpenAI(model="gpt-4o", temperature=0)

@tool
def get_order_status(order_id: str) -> str:
    """Get the current status of an order."""
    # Simulate a changing state
    return "shipped" if datetime.now().second % 2 == 0 else "pending"

class PlanAndSolveAgent:
    def __init__(self, llm, tools, max_plan_age_seconds=5):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_plan_age = timedelta(seconds=max_plan_age_seconds)
        self.plan = None
        self.plan_created_at = None

    def generate_plan(self, task: str) -> list[str]:
        prompt = f"Generate a step-by-step plan to accomplish this task. Return a JSON list of strings. Task: {task}"
        response = self.llm.invoke(prompt)
        # Parse the JSON response; add schema validation
        try:
            plan = json.loads(response.content)
            if not isinstance(plan, list):
                raise ValueError("Plan must be a list")
        except (json.JSONDecodeError, ValueError) as e:
            print(f"Plan parsing failed: {e}. Falling back to single step.")
            plan = [f"Complete task: {task}"]
        self.plan = plan
        self.plan_created_at = datetime.now()
        return plan

    def is_plan_stale(self) -> bool:
        return datetime.now() - self.plan_created_at > self.max_plan_age

    def execute_step(self, step: str) -> str:
        # Parse step to extract tool call
        if "check order" in step.lower():
            return self.tools["get_order_status"].invoke({"order_id": "ORD-123"})
        return f"Executed: {step}"

    def run(self, task: str):
        self.generate_plan(task)
        for i, step in enumerate(self.plan):
            if self.is_plan_stale():
                print(f"Plan is stale (age > {self.max_plan_age}). Re-planning.")
                self.generate_plan(task)
            observation = self.execute_step(step)
            print(f"Step {i}: {step} -> {observation}")
            # Check for error: if observation indicates failure, re-plan
            if "error" in observation.lower() or "failed" in observation.lower():
                print("Step failed. Re-planning from current state.")
                self.generate_plan(f"Recover from failure at step {i}. Current state: {observation}. Original task: {task}")

agent = PlanAndSolveAgent(llm, [get_order_status])
agent.run("Process order ORD-123")

import json
from langchain_openai import ChatOpenAI

# Use two models: one for generation (expensive), one for evaluation (cheap)
gen_llm = ChatOpenAI(model="gpt-4o", temperature=0.7, max_tokens=200)
eval_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, max_tokens=50)

class TreeOfThought:
    def __init__(self, gen_llm, eval_llm, max_nodes=50, branching_factor=3, max_depth=5):
        self.gen_llm = gen_llm
        self.eval_llm = eval_llm
        self.max_nodes = max_nodes
        self.branching_factor = branching_factor
        self.max_depth = max_depth
        self.nodes_visited = 0

    def generate_thoughts(self, state: str, num_thoughts: int) -> list[str]:
        prompt = f"Given the current state: '{state}', generate {num_thoughts} distinct next steps. Return as a JSON list of strings."
        response = self.gen_llm.invoke(prompt)
        try:
            thoughts = json.loads(response.content)[:num_thoughts]
        except:
            thoughts = [f"Fallback step for state: {state}"]
        return thoughts

    def evaluate_thought(self, thought: str) -> float:
        prompt = f"Rate the promise of this thought on a scale of 0 to 1. Thought: '{thought}'. Return only the number."
        response = self.eval_llm.invoke(prompt)
        try:
            return float(response.content.strip())
        except:
            return 0.5

    def search(self, initial_state: str) -> str:
        from heapq import heappush, heappop
        # Priority queue: (-score, depth, state)
        queue = []
        heappush(queue, (0, 0, initial_state))
        best_state = initial_state
        best_score = 0

        while queue and self.nodes_visited < self.max_nodes:
            neg_score, depth, state = heappop(queue)
            self.nodes_visited += 1

            if depth >= self.max_depth:
                continue

            # Generate and evaluate branches
            thoughts = self.generate_thoughts(state, self.branching_factor)
            for thought in thoughts:
                score = self.evaluate_thought(thought)
                if score > best_score:
                    best_score = score
                    best_state = thought
                # Push with negative score for max-heap behavior
                heappush(queue, (-score, depth + 1, thought))

        print(f"Visited {self.nodes_visited} nodes (budget: {self.max_nodes})")
        return best_state

tot = TreeOfThought(gen_llm, eval_llm, max_nodes=30)  # Aggressive budget
result = tot.search("I need to debug a production issue: p99 latency spike")
print(f"Best thought: {result}")

# Example: simple pipeline vs. ReAct agent for a deterministic task

# Simple pipeline (no planning) — 50 lines, 100ms latency
import requests

def process_refund(user_id: str, amount: float) -> str:
    user = requests.get(f"https://api.example.com/users/{user_id}").json()
    if user["balance"] < amount:
        return "Insufficient balance"
    txn = requests.post("https://api.example.com/refunds", json={"user_id": user_id, "amount": amount})
    return txn.json()["status"]

# ReAct agent (planning) — 200 lines, 2s latency, $0.05 per call
# from langchain.agents import ... (not shown for brevity)

# Decision helper
from enum import Enum

class TaskComplexity(Enum):
    DETERMINISTIC = 1  # Use pipeline
    CONDITIONAL = 2    # Use CoT or ReAct
    EXPLORATORY = 3    # Use ToT or Reflexion

def classify_task(task_description: str) -> TaskComplexity:
    # Simple heuristic: if the task has 'if' conditions and multiple tools, use planning
    if "if" in task_description and "tool" in task_description:
        return TaskComplexity.CONDITIONAL
    if "explore" in task_description or "hypothesis" in task_description:
        return TaskComplexity.EXPLORATORY
    return TaskComplexity.DETERMINISTIC

# Use this to decide which implementation to deploy
print(classify_task("Refund a user if balance is sufficient"))  # DETERMINISTIC

import hashlib
import time
from functools import lru_cache
from typing import Any

# Pattern 1: Caching the planning step
class PlanCache:
    def __init__(self, maxsize=1000):
        self.cache = {}
        self.maxsize = maxsize

    def _make_key(self, task: str, history: str) -> str:
        return hashlib.md5((task + history).encode()).hexdigest()

    def get(self, task: str, history: str) -> Any | None:
        key = self._make_key(task, history)
        return self.cache.get(key)

    def set(self, task: str, history: str, plan: Any):
        key = self._make_key(task, history)
        if len(self.cache) >= self.maxsize:
            # Evict oldest (simple FIFO)
            self.cache.pop(next(iter(self.cache)))
        self.cache[key] = plan

# Pattern 2: Circuit Breaker
class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise Exception("Circuit breaker is open")

        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise e

# Usage
plan_cache = PlanCache(maxsize=1000)
circuit_breaker = CircuitBreaker()

def get_plan(task: str, history: str) -> Any:
    cached = plan_cache.get(task, history)
    if cached:
        return cached
    # Simulate LLM call
    plan = {"steps": ["step1", "step2"]}
    plan_cache.set(task, history, plan)
    return plan

# Wrap with circuit breaker
result = circuit_breaker.call(get_plan, "investigate fraud", "history...")
print(result)

# Mistake 1: Not handling tool errors
# Fix: wrap tool with retry logic
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
def safe_tool_call(tool_func, **kwargs):
    try:
        return tool_func(**kwargs)
    except Exception as e:
        return f"Error: {str(e)}"  # Return error as observation, don't crash

# Mistake 2: Same LLM for planning and evaluation
# Fix: separate models
from langchain_openai import ChatOpenAI
plan_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
eval_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Mistake 3: Not logging the trace
# Fix: add a callback that logs every step
from langchain.callbacks import StdOutCallbackHandler
handler = StdOutCallbackHandler()  # Logs to stdout; use a file handler in production

# Mistake 4: Prompt injection in tool inputs
# Fix: sanitize inputs
import re
def sanitize_input(user_input: str) -> str:
    # Remove common injection patterns
    return re.sub(r"ignore all previous instructions|system prompt|you are an ai", "", user_input, flags=re.IGNORECASE)

# Production decision helper
from enum import Enum

class PlanningStrategy(Enum):
    REACT = "react"
    PLAN_AND_SOLVE = "plan_and_solve"
    TREE_OF_THOUGHT = "tree_of_thought"
    REFLEXION = "reflexion"
    NONE = "none"

def select_strategy(task_type: str, latency_budget_ms: int, cost_budget_per_task: float) -> PlanningStrategy:
    """
    Select the best planning strategy based on task characteristics and budget.
    
    Args:
        task_type: 'deterministic', 'conditional', 'exploratory'
        latency_budget_ms: maximum acceptable latency in milliseconds
        cost_budget_per_task: maximum acceptable cost in dollars
    """
    if task_type == "deterministic":
        return PlanningStrategy.NONE
    
    if task_type == "conditional":
        if latency_budget_ms < 2000:
            return PlanningStrategy.PLAN_AND_SOLVE
        else:
            return PlanningStrategy.REACT
    
    if task_type == "exploratory":
        if cost_budget_per_task < 0.50:
            return PlanningStrategy.REACT  # Cheaper than ToT
        else:
            return PlanningStrategy.TREE_OF_THOUGHT
    
    return PlanningStrategy.REACT  # Default

# Example usage
strategy = select_strategy("exploratory", latency_budget_ms=5000, cost_budget_per_task=2.0)
print(f"Selected strategy: {strategy.value}")  # tree_of_thought

import json
import logging
from datetime import datetime
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

# Structured logging for planning traces
logger = logging.getLogger("planning")
logger.setLevel(logging.INFO)
handler = logging.FileHandler("planning_trace.log")
handler.setFormatter(logging.Formatter(json.dumps({
    "timestamp": "%(asctime)s",
    "level": "%(levelname)s",
    "message": "%(message)s"
})))
logger.addHandler(handler)

# OpenTelemetry tracing
tracer = trace.get_tracer(__name__)

def trace_planning_step(step_name: str, func):
    def wrapper(*args, **kwargs):
        with tracer.start_as_current_span(step_name) as span:
            try:
                result = func(*args, **kwargs)
                span.set_status(Status(StatusCode.OK))
                span.set_attribute("step.result", str(result)[:200])
                return result
            except Exception as e:
                span.set_status(Status(StatusCode.ERROR, str(e)))
                raise
    return wrapper

# Plan quality scorer (uses a separate LLM call)
def score_plan(plan: list[str], task: str) -> float:
    from langchain_openai import ChatOpenAI
    eval_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = f"Rate the quality of this plan for the task '{task}' on a scale of 0 to 1. Return only the number. Plan: {json.dumps(plan)}"
    response = eval_llm.invoke(prompt)
    try:
        score = float(response.content.strip())
        return score
    except:
        return 0.5

# Usage
@trace_planning_step("generate_plan")
def generate_plan(task: str) -> list[str]:
    # ... planning logic
    plan = ["step1", "step2"]
    score = score_plan(plan, task)
    logger.info(json.dumps({"event": "plan_generated", "task": task, "plan": plan, "score": score}))
    if score < 0.8:
        logger.warning(f"Low-quality plan (score={score}) for task: {task}")
    return plan