managing-agentic-loops

name	managing-agentic-loops
description	ReAct pattern implementation, reflection and self-correction, planning Use when this capability is needed.
metadata	{"author":"gitwalter"}

Agentic Loops

ReAct pattern implementation, reflection and self-correction, planning and task decomposition, iterative refinement patterns

Implement agentic reasoning loops - ReAct pattern, reflection, planning, and iterative refinement for autonomous agent behavior.

Process

1. Review the task requirements.
2. Apply the skill's methodology.
3. Validate the output against the defined criteria.

Step 1: Basic ReAct Pattern

Implement the core ReAct loop: Reason → Act → Observe → Repeat:

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import List

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

@tool
def search_knowledge_base(query: str) -> str:
    """Search internal knowledge base for information.

    Args:
        query: Search query
    """
    # Implementation
    return f"Knowledge base results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression.

    Args:
        expression: Math expression like '2 + 2'
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

tools = [search_knowledge_base, calculate]
llm_with_tools = llm.bind_tools(tools)

async def react_loop(user_query: str, max_iterations: int = 10) -> str:
    """Basic ReAct pattern implementation."""
    messages: List = [HumanMessage(content=user_query)]
    iteration = 0

    while iteration < max_iterations:
        # REASON: LLM generates reasoning and decides on actions
        response = await llm_with_tools.ainvoke(messages)
        messages.append(response)

        # Check if agent is done
        if not response.tool_calls:
            return response.content

        # ACT: Execute tool calls
        tool_map = {t.name: t for t in tools}
        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_id = tool_call["id"]

            if tool_name in tool_map:
                try:
                    # OBSERVE: Get tool result
                    if asyncio.iscoroutinefunction(tool_map[tool_name].func):
                        result = await tool_map[tool_name].ainvoke(tool_args)
                    else:
                        result = tool_map[tool_name].invoke(tool_args)

                    messages.append(ToolMessage(
                        content=str(result),
                        tool_call_id=tool_id
                    ))
                except Exception as e:
                    messages.append(ToolMessage(
                        content=f"Error: {str(e)}",
                        tool_call_id=tool_id
                    ))

        iteration += 1

    return "Maximum iterations reached."

# Usage
result = await react_loop("What is 15 * 23 and search for information about Python?")

Step 2: Reflection and Self-Correction

Add reflection step to evaluate and correct actions:

from langchain_core.prompts import ChatPromptTemplate

class ReflectiveAgent:
    """Agent with reflection and self-correction capabilities."""

    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.llm_with_tools = llm.bind_tools(tools)

        # Reflection prompt
        self.reflection_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a reflective agent. After taking actions, evaluate:
1. Did the actions achieve the goal?
2. Were there any errors or issues?
3. What should be done differently?
4. Should we continue or stop?"""),
            ("user", "{history}")
        ])

    async def reflect(self, messages: List, goal: str) -> dict:
        """Reflect on current progress toward goal."""
        history = "\n".join([
            f"{msg.__class__.__name__}: {msg.content}"
            for msg in messages[-10:]  # Last 10 messages
        ])

        reflection = await self.reflection_prompt.ainvoke({
            "history": f"Goal: {goal}\n\nConversation:\n{history}"
        })

        # Parse reflection
        content = reflection.content.lower()
        should_continue = "continue"  content    content
        errors_found =   content    content

         {
            : reflection.content,
            : should_continue,
            : errors_found
        }

      () -> :
        
        messages = [HumanMessage(content=user_query)]
        iteration = 

         iteration < max_iterations:
            
            response =  .llm_with_tools.ainvoke(messages)
            messages.append(response)

              response.tool_calls:
                
                reflection =  .reflect(messages, user_query)
                 reflection[]  reflection[]:
                    
                    correction_msg = HumanMessage(
                        content=
                    )
                    messages.append(correction_msg)
                    
                 response.content

            
            tool_map = {t.name: t  t  .tools}
             tool_call  response.tool_calls:
                tool_name = tool_call[]
                tool_args = tool_call[]
                tool_id = tool_call[]

                 tool_name  tool_map:
                    :
                         asyncio.iscoroutinefunction(tool_map[tool_name].func):
                            result =  tool_map[tool_name].ainvoke(tool_args)
                        :
                            result = tool_map[tool_name].invoke(tool_args)
                        messages.append(ToolMessage(
                            content=(result),
                            tool_call_id=tool_id
                        ))
                     Exception  e:
                        messages.append(ToolMessage(
                            content=,
                            tool_call_id=tool_id
                        ))

            
             iteration %  == :
                reflection =  .reflect(messages, user_query)
                  reflection[]:
                     

            iteration += 

         


agent = ReflectiveAgent(llm, tools)
result =  agent.execute()

Step 3: Planning and Task Decomposition

Break complex tasks into subtasks:

from pydantic import BaseModel, Field
from typing import List

class Subtask(BaseModel):
    """Represents a subtask in a plan."""
    id: int
    description: str
    dependencies: List[int] = Field(default_factory=list)
    status: str = "pending"  # pending, in_progress, completed, failed

class Planner:
    """Agent that plans and decomposes tasks."""

    def __init__(self, llm):
        self.llm = llm
        self.planning_prompt = ChatPromptTemplate.from_messages([
            ("system", """Break down the task into subtasks. For each subtask, identify:
1. What needs to be done
2. Dependencies on other subtasks
3. Order of execution

Return subtasks as a structured list."""),
            ("user", "{task}")
        ])

    async def create_plan(self, task: str) -> List[Subtask]:
        """Decompose task into subtasks."""
        response = await self.planning_prompt.ainvoke({"task": task})

        # Parse response into subtasks (simplified - in practice use structured output)
        # For now, return example structure
        subtasks = [
            Subtask(=, description=, dependencies=[]),
            Subtask(=, description=, dependencies=[]),
            Subtask(=, description=, dependencies=[]),
        ]

         subtasks

      () -> :
        
        plan =  .create_plan(task)
        results = {}
        llm_with_tools = .llm.bind_tools(tools)

         subtask  plan:
            
             (results.get(dep_id) ==   dep_id  subtask.dependencies):
                subtask.status = 
                results[subtask.] = 
                

            subtask.status = 

            
            query = 
            messages = [HumanMessage(content=query)]

            iteration = 
             iteration < :  
                response =  llm_with_tools.ainvoke(messages)
                messages.append(response)

                  response.tool_calls:
                    results[subtask.] = response.content
                    subtask.status = 
                    

                
                tool_map = {t.name: t  t  tools}
                 tool_call  response.tool_calls:
                    tool_name = tool_call[]
                    tool_args = tool_call[]
                    tool_id = tool_call[]

                     tool_name  tool_map:
                        :
                             asyncio.iscoroutinefunction(tool_map[tool_name].func):
                                result =  tool_map[tool_name].ainvoke(tool_args)
                            :
                                result = tool_map[tool_name].invoke(tool_args)
                            messages.append(ToolMessage(
                                content=(result),
                                tool_call_id=tool_id
                            ))
                         Exception  e:
                            messages.append(ToolMessage(
                                content=,
                                tool_call_id=tool_id
                            ))

                iteration += 

             iteration >= :
                subtask.status = 
                results[subtask.] = 

        
         


planner = Planner(llm)
result =  planner.execute_plan(, tools)

Step 4: Iterative Refinement Pattern

Refine outputs through multiple iterations:

class RefinementAgent:
    """Agent that iteratively refines its output."""

    def __init__(self, llm):
        self.llm = llm
        self.refinement_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an agent that improves outputs iteratively.
Given the current output and feedback, refine it to be better.
Continue refining until the output meets quality standards."""),
            ("user", """Original request: {request}
Current output: {current_output}
Feedback: {feedback}
Iteration: {iteration}

Refine the output based on the feedback.""")
        ])

    async def refine(
        self,
        request: str,
        initial_output: str = None,
        max_iterations: int = 5,
        quality_threshold: float = 0.8
    ) -> str:
        """Iteratively refine output."""
        current_output = initial_output or ""
        iteration = 0

        while iteration < max_iterations:
            # Generate or refine
            if iteration == 0 and not initial_output:
                # First iteration - generate initial output
                response = await self.llm.ainvoke(request)
                current_output = response.content
            else:
                # Refinement iteration
                feedback =  ._evaluate_quality(request, current_output)

                 feedback[] >= quality_threshold:
                     current_output

                response =  .refinement_prompt.ainvoke({
                    : request,
                    : current_output,
                    : feedback[],
                    : iteration + 
                })
                current_output = response.content

            iteration += 

         current_output

      () -> :
        
        eval_prompt = ChatPromptTemplate.from_messages([
            (, ),
            (, )
        ])

        response =  .llm.ainvoke(eval_prompt.(request=request, output=output))

        
        content = response.content.lower()
        score =   
           content    content:
            score = 
           content:
            score = 
           content:
            score = 

         {
            : score,
            : response.content
        }


refiner = RefinementAgent(llm)
refined =  refiner.refine(
    ,
    max_iterations=
)

Step 5: Parallel Tool Execution

Execute multiple tools in parallel for efficiency:

import asyncio
from typing import List, Dict

async def parallel_react_loop(user_query: str, tools: List, max_iterations: int = 10) -> str:
    """ReAct loop with parallel tool execution."""
    llm_with_tools = llm.bind_tools(tools)
    messages = [HumanMessage(content=user_query)]
    iteration = 0

    while iteration < max_iterations:
        response = await llm_with_tools.ainvoke(messages)
        messages.append(response)

        if not response.tool_calls:
            return response.content

        # Execute all tool calls in parallel
        tool_map = {t.name: t for t in tools}
        tool_tasks = []

        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_id = tool_call["id"]

            if tool_name in tool_map:
                tool = tool_map[tool_name]

                async def execute_tool(tool, args, tool_id):
                    try:
                        if asyncio.iscoroutinefunction(tool.func):
                            result = await tool.ainvoke(args)
                        else:
                            result = tool.invoke(args)
                        return ToolMessage(
                            content=(result),
                            tool_call_id=tool_id
                        )
                     Exception  e:
                         ToolMessage(
                            content=,
                            tool_call_id=tool_id
                        )

                tool_tasks.append(execute_tool(tool, tool_args, tool_id))

        
        tool_results =  asyncio.gather(*tool_tasks)
        messages.extend(tool_results)

        iteration += 

     

Step 6: Conditional Loop Control

Add conditions to control loop behavior:

class ConditionalAgent:
    """Agent with conditional loop control."""

    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.llm_with_tools = llm.bind_tools(tools)

    async def execute(
        self,
        user_query: str,
        stop_conditions: List[str] = None,
        max_iterations: int = 10,
        min_confidence: float = 0.7
    ) -> str:
        """Execute with conditional stopping."""
        messages = [HumanMessage(content=user_query)]
        iteration = 0
        stop_conditions = stop_conditions or []

        while iteration < max_iterations:
            response = await self.llm_with_tools.ainvoke(messages)
            messages.append(response)

            # Check stop conditions
            content_lower = response.content.lower()
            if any(condition.lower() in content_lower for condition in stop_conditions):
                return f"Stopped due to condition. {response.content}"

            if not response.tool_calls:
                # Check confidence (simplified)
                confidence = await self._estimate_confidence(response.content)
                 confidence >= min_confidence:
                     response.content
                :
                    
                    messages.append(HumanMessage(
                        content=
                    ))
                    

            
            tool_map = {t.name: t  t  .tools}
             tool_call  response.tool_calls:
                tool_name = tool_call[]
                tool_args = tool_call[]
                tool_id = tool_call[]

                 tool_name  tool_map:
                    :
                         asyncio.iscoroutinefunction(tool_map[tool_name].func):
                            result =  tool_map[tool_name].ainvoke(tool_args)
                        :
                            result = tool_map[tool_name].invoke(tool_args)
                        messages.append(ToolMessage(
                            content=(result),
                            tool_call_id=tool_id
                        ))
                     Exception  e:
                        messages.append(ToolMessage(
                            content=,
                            tool_call_id=tool_id
                        ))

            iteration += 

         

      () -> :
        
        uncertainty_words = [, , , ]
        content_lower = content.lower()

         (word  content_lower  word  uncertainty_words):
             
         


agent = ConditionalAgent(llm, tools)
result =  agent.execute(
    ,
    stop_conditions=[, ],
    min_confidence=
)

### Step 7: Sequential Thinking Integration (New)

Use the `sequential-thinking` MCP server for complex problem solving:

```python
# Decompose a complex problem
response = await client.chat.completions.create(
    messages=[{"role": "user", "content": "Design a scalable microservices architecture for a banking app"}],
    tools=[{
        "type": "mcp",
        "name": "sequential-thinking",
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-sequential-thinking"]
    }]
)

Agentic Loop Patterns

| Pattern | Description | Use Case | ||-|-| | ReAct | Reason → Act → Observe → Repeat | General agentic tasks | | Reflection | Act → Reflect → Correct → Repeat | Quality-critical tasks | | Planning | Plan → Decompose → Execute → Aggregate | Complex multi-step tasks | | Refinement | Generate → Evaluate → Refine → Repeat | Content generation | | Parallel | Act → Execute tools in parallel → Observe | Performance-critical | | Conditional | Act → Check conditions → Continue/Stop | Controlled execution |

Best Practices

• Set appropriate max_iterations to prevent infinite loops
• Implement proper error handling for tool calls
• Use reflection for quality-critical tasks
• Decompose complex tasks into subtasks
• Execute independent tools in parallel
• Add stop conditions for controlled execution
• Monitor token usage in long-running loops
• Log loop iterations for debugging
• Use structured outputs for planning
• Implement confidence thresholds

Anti-Patterns

Anti-Pattern	Fix
No iteration limit	Set max_iterations
Sequential tool execution	Use asyncio.gather for parallel execution
No error handling	Wrap tool calls in try/except
No reflection	Add reflection step for quality
Ignoring tool errors	Report errors back to agent
No planning	Decompose complex tasks
Infinite loops	Add stop conditions
No confidence checking	Evaluate response quality
Synchronous execution	Use async/await throughout
No logging	Log iterations and decisions

Related

• Knowledge: {directories.knowledge}/agentic-loop-patterns.json
• Skill: anthropic-patterns
• Skill: tool-usage
• Skill: using-langchain
• Skill: langgraph-agent-building

When to Use

This skill should be used when strict adherence to the defined process is required.

Prerequisites

• Basic understanding of the agent factory context.
• Access to the necessary tools and resources.

---

Converted and distributed by TomeVault — claim your Tome and manage your conversions.