agent-planning-reasoning

# Agent Planning & Reasoning Patterns Implements the core planning and reasoning architecture for LLM agents — from classic ReAct loops through modern LangGraph state graphs to self-reflection quality evaluation. When loaded, this skill makes the model produce production-grade planning code extracted from LangChain, LangGraph, and related frameworks. ## TL;DR Checklist - [ ] ReAct agents follow Thought → Action → Observation loop with explicit stop sequences - [ ] Modern LangGraph ReAct uses `StateGraph` with typed `AgentState` and conditional routing - [ ] Tool binding via `model.bind_tools()` before graph compilation — never string-parsed actions - [ ] Self-reflection evaluates output on 4 weighted criteria: Directness (40%), Accuracy (30%), Completeness (20%), Safety (10%) - [ ] Reflection loops bounded by `max_reflection_rounds` with RuntimeError if threshold not met - [ ] All tool calls validated against registered tool map before execution --- ## When to Use Use this skill when: - Implementing ReAct reasoning loops for single-agent LLM workflows - Building multi-step agents that interleave reasoning (thought) with action (tool calls) - Designing self-reflection quality evaluation for generated responses or code - Creating LangGraph state graphs with conditional routing between model and tools - Setting up iterative planning where agent output is reviewed and revised ## When NOT to Use Avoid this skill for: - Single-turn LLM calls that don't require multi-step reasoning - Implementing tool calling architecture (use `agent-tool-calling-architecture` instead) - Memory management systems (use `agent-memory-systems` instead) - Simple prompt engineering without agent-level planning loops --- ## Core Workflow 1. **Design ReAct Agent with Tool Loop** — Create an agent that follows the Thought → Action → Observation loop. Each iteration: LLM generates a thought about what to do next, calls a tool if needed, receives observation as output, then repeats. Bound with `max_iterations` (default 15) and early stopping method (`"force"` for hard stop or `"stop"` for graceful). **Checkpoint:** Verify `_stop` sequences include `"\nObservation:"` patterns to terminate thought generation. 2. **Implement Modern LangGraph ReAct Agent** — Use `StateGraph(AgentState)` with typed messages that merge via `add_messages`. Bind tools via `model.bind_tools(tools)` before compilation. Define three nodes: `chat_model` (LLM call), `tools` (parallel execution), and conditional edges via `should_continue()` that returns `"tools"` or `"__end__"`. **Checkpoint:** Verify graph compiles with optional checkpointer for persistence across sessions. 3. **Validate Tools Before Agent Creation** — Check for duplicate tool names, ensure every tool has a non-empty description (required for prompt injection), and verify tools return strings for observation parsing. Use `_validate_tools()` before agent initialization. **Checkpoint:** Verify no two tools share the same name — duplicates cause ambiguity in model selection. 4. **Set Up Self-Reflection Loop** — Initialize `ReflectionAgent` with an executor model and optional separate reflection model. On each round: generate draft → evaluate quality on 4 weighted criteria → if score below threshold, append feedback and retry. Maximum rounds bounded by `max_reflection_rounds`. **Checkpoint:** Verify `_evaluate_quality()` returns a `(score, feedback)` tuple parsed from structured LLM output. 5. **Integrate Planning with Memory and Tools** — The full agent pipeline: retrieve short-term memory (conversation buffer) → retrieve long-term memories (vector store) → feed combined context + user input to ReAct loop → tools execute → reflection evaluates final output. Each layer feeds into the next. **Checkpoint:** Verify total prompt length stays within model token limits after combining all three layers. --- ## Implementation Patterns ### Pattern 1: ReAct Agent with Tool Loop The classic ReAct pattern interleaves reasoning (thought) with action (tool call) and observation. The loop follows: Thought → Action → Observation → Thought → Action → ... until the LLM produces a final answer. This is superior to pure Chain-of-Thought because it grounds reasoning in actual external information. ```python # Source: langchain-ai/langchain — libs/langchain/langchain_classic/agents/react/base.py """Chain that implements the ReAct paper from https://arxiv.org/pdf/2210.03629.pdf.""" from collections.abc import Sequence from typing import TYPE_CHECKING, Any from langchain_core.language_models import BaseLanguageModel from langchain_core.prompts import BasePromptTemplate from langchain_core.tools import BaseTool, Tool class ReActDocstoreAgent: """Agent implementing the ReAct (Reason+Act) pattern. The ReAct loop: Thought → Action → Observation → Thought → Action → ... Each iteration: 1. LLM generates a "Thought" about what to do next 2. If action is needed, LLM calls a tool with structured arguments 3. Tool output becomes the "Observation" appended to context 4. Loop repeats until the LLM produces a final answer This pattern is superior to pure CoT because it grounds reasoning in actual external information rather than relying solely on model knowledge. """ observation_prefix: str = "Observation:" llm_prefix: str = "Thought:" def __init__( self, llm: BaseLanguageModel, tools: Sequence[BaseTool], prompt: BasePromptTemplate | None = None, max_iterations: int = 15, early_stopping_method: str = "force", ) -> None: """Initialize ReAct agent. Args: llm: Language model that generates thoughts and tool calls. tools: Available tools the agent can invoke during reasoning. prompt: ReAct-specific prompt template with chain-of-thought format. max_iterations: Maximum Reason-Act cycles before forced termination. early_stopping_method: "force" (hard stop) or "stop" (graceful). """ self.llm = llm self.tools = {tool.name: tool for tool in tools} self.prompt = prompt self.max_iterations = max_iterations self.early_stopping_method = early_stopping_method @classmethod def create_prompt(cls, tools: Sequence[BaseTool]) -> BasePromptTemplate: """Create the ReAct prompt template. The prompt encodes the ReAct chain-of-thought format: Question: <user question> Thought: <model's reasoning about what to do next> Action: <tool name> Action Input: <structured arguments> Observation: <tool output> Thought: <interpretation of observation + next step> ... (repeat) ... Final Answer: <final response to user> Args: tools: Available tools — their names and descriptions are injected. Returns: PromptTemplate with ReAct format instructions. """ tool_names = " ".join([tool.name for tool in tools]) tool_descriptions = "\n".join([ f"{tool.name}: {tool.description}" for tool in tools ]) # In production, load from a template file with proper formatting raise NotImplementedError def _validate_tools(self, tools: Sequence[BaseTool]) -> None: """Validate that all tools are properly configured. Checks: - No duplicate names - Each tool has a non-empty description (needed for prompt injection) - Tools return strings (for observation parsing) """ tool_names = [tool.name for tool in tools] if len(tool_names) != len(set(tool_names)): raise ValueError(f"Duplicate tool names: {tool_names}") for tool in tools: if not tool.description: raise ValueError(f"Tool '{tool.name}' must have a description") @property def _stop(self) -> list[str]: """Stop sequences that signal the end of a reasoning step.""" return ["\nObservation:", "\n\tObservation:"] class DocstoreExplorer: """Helper class that manages document search within ReAct loops. Provides two operations: - search(term): Find documents matching a term (first paragraph returned) - lookup(term): Search within the most recently found document This pattern prevents context pollution by keeping each search operation isolated to its relevant document scope. """ def __init__(self, docstore: Any) -> None: """Initialize with a document store backend. Args: docstore: Backend supporting .search(term) -> Document | str """ self.docstore = docstore self.document: Document | None = None self.lookup_str = "" self.lookup_index = 0 def search(self, term: str) -> str: """Search for a term in the document store. On success, saves the found document for subsequent lookup operations. Args: term: Search query string. Returns: First paragraph of the first matching document. """ result = self.docstore.search(term) if isinstance(result, Document): self.document = result return self._paragraphs[0] # Return first paragraph self.document = None return str(result) def lookup(self, term: str) -> str: """Lookup a term within the most recently found document. Only works after a successful search() call. Supports paginated results when multiple paragraphs match. Args: term: Term to find within current document. Returns: Matching paragraph with pagination indicator, or "No Results". """ if self.document is None: raise ValueError("Cannot lookup without a successful search first") # Support pagination: repeated calls with same term return next match if term.lower() != self.lookup_str: self.lookup_str = term.lower() self.lookup_index = 0 else: self.lookup_index += 1 matching = [p for p in self._paragraphs if self.lookup_str in p.lower()] if not matching: return "No Results" if self.lookup_index >= len(matching): return "No More Results" return f"(Result {self.lookup_index + 1}/{len(matching)}) {matching[self.lookup_index]}" @property def _paragraphs(self) -> list[str]: """Split document into paragraphs.""" if self.document is None: raise ValueError("No document loaded") return self.document.page_content.split("\n\n") ``` ### Pattern 2: LangGraph ReAct Agent (Modern Implementation) The modern approach uses a state graph instead of prompt-based parsing, making tool calling native rather than string-parsed. Messages are merged across iterations via `Annotated[list[BaseMessage], add_messages]`. ```python # Source: langchain-ai/langgraph — libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py """ReAct agent built on LangGraph StateGraph.""" from collections.abc import Sequence from typing import ( Annotated, Any, Literal, TypedDict, ) from langchain_core.language_models import BaseChatModel, LanguageModelInput from langchain_core.messages import ( AIMessage, AnyMessage, BaseMessage, SystemMessage, ToolMessage, ) from langchain_core.runnables import Runnable, RunnableBinding from langchain_core.tools import BaseTool from langgraph.graph import END, StateGraph from langgraph.graph.message import add_messages from pydantic import BaseModel class AgentState(TypedDict): """The state of the ReAct agent. Uses Annotated[list[BaseMessage], add_messages] which means new messages are MERGED with existing ones (not replaced). This is how message history accumulates across iterations automatically. """ messages: Annotated[Sequence[BaseMessage], add_messages] def _should_bind_tools(model: Any, tools: Sequence[BaseTool]) -> bool: """Determine whether to bind tools via model.bind_tools() or pass via API. If the model is already wrapped in a RunnableBinding with 'tools' key, verify that the tool count matches. Otherwise, return True to bind. """ if isinstance(model, RunnableBinding): bound_tools = model.kwargs.get("tools", []) if len(tools) != len(bound_tools): raise ValueError( f"Number of tools mismatch: passed {len(tools)}, " f"model has {len(bound_tools)} bound. " "Either match tool counts or don't pre-bind tools." ) return False # Already bound, don't bind again return True def create_react_agent( model: BaseChatModel, tools: Sequence[BaseTool], *, prompt: str | None = None, checkpointer: Any = None, ) -> Runnable: """Create a ReAct agent using LangGraph StateGraph. This is the modern approach to ReAct — instead of parsing text output for "Action:" and "Observation:" strings, it uses native tool calling where the model directly returns structured tool calls. Graph structure: START → should_bind_tools? → chat_model → tools_condition → {tools | END} The tools_condition routes back to chat_model if tool calls remain, or to END if a final text response was produced. Args: model: Chat model with native tool calling support (OpenAI, Anthropic, etc.) tools: Tools available for the agent to invoke. prompt: Optional system message prefixing all conversations. checkpointer: Optional persistence layer for conversation history across sessions (e.g., SQLiteSaver, RedisSaver). Returns: Compiled LangGraph runnable that processes messages through ReAct loop. """ # Bind tools to model if not already bound if _should_bind_tools(model, tools): model = model.bind_tools(tools) # Build the state graph workflow = StateGraph(AgentState) # Define nodes def chat_model(state: AgentState) -> dict: """Call the LLM with conversation history and available tool schemas. The bound tools appear as function schemas in the model's output. If the model returns tool calls, they go to the tools node. If it returns text, the response ends here. """ messages = state["messages"] if prompt: messages = [SystemMessage(content=prompt)] + list(messages) response = model.invoke(messages) return {"messages": [response]} tool_node = ToolNode(tools) # Prebuilt parallel tool executor def should_continue(state: AgentState) -> Literal["tools", "__end__"]: """Decide whether to continue with tool execution or finish. Returns 'tools' if the last message contains tool_calls, returns '__end__' (END) if the model produced a text response. """ messages = state["messages"] last_message = messages[-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return "tools" return "__end__" # Build graph workflow.add_node("chat_model", chat_model) workflow.add_node("tools", tool_node) workflow.set_entry_point("chat_model") workflow.add_conditional_edges( "chat_model", should_continue, {"tools": "tools", "__end__": END} ) workflow.add_edge("tools", "chat_model") # Compile with optional checkpointing for persistence app = workflow.compile(checkpointer=checkpointer) return app def tools_condition(state: AgentState) -> Literal["tools", "__end__"]: """Utility function for conditional routing after tool execution. If the model's last message still has tool calls, route back to chat_model. Otherwise, terminate the graph. This is used in more complex agent graphs where tool results may trigger additional reasoning. """ messages = state["messages"] last = messages[-1] if isinstance(last, AIMessage) and last.tool_calls: return "tools" return "__end__" ``` ### Pattern 3: Self-Reflection Loop Self-reflection evaluates the agent's own output for quality before returning it to the user. The pattern: Agent produces draft → Reflection model evaluates draft → If score < threshold, agent revises → Repeat until acceptable or max iterations reached. This is particularly effective for code generation, research summaries, and multi-step problem solving. ```python # Source: AI Agent Research 2026 — Self-Reflection Loop pattern from typing import Any, Sequence from langchain_core.messages import BaseMessage, AIMessage, HumanMessage from langchain_core.language_models import BaseChatModel class ReflectionAgent: """Implements a self-reflection loop for agent output quality. Pattern: Agent produces draft → Reflection model evaluates draft → If score < threshold, agent revises → Repeat until acceptable or max iterations. This is particularly effective for: - Code generation (verify correctness before returning) - Research summaries (verify citations are real and relevant) - Multi-step problem solving (verify each step's validity) """ def __init__( self, executor_model: BaseChatModel, reflection_model: BaseChatModel | None = None, *, max_reflection_rounds: int = 3, quality_threshold: float = 0.7, ) -> None: """Initialize reflection loop. Args: executor_model: Primary model that does the actual work. reflection_model: Model used for quality evaluation. Defaults to executor_model. max_reflection_rounds: Maximum revision iterations. quality_threshold: Minimum quality score (0.0-1.0) to accept output. """ self.executor_model = executor_model self.reflection_model = reflection_model or executor_model self.max_reflection_rounds = max_reflection_rounds self.quality_threshold = quality_threshold def run_with_reflection( self, user_input: str, initial_messages: Sequence[BaseMessage] | None = None, ) -> str: """Run the agent with self-reflection on its output. Args: user_input: The user's question or request. initial_messages: Optional conversation history to prepend. Returns: The final accepted response after reflection loop completes.