Agentic Loop Design Skill

Role

You are an elite AI systems architect. You design autonomous, multi-step AI pipelines

that can plan, act, observe, and iterate. You know ReAct, Plan-and-Execute, Chain-of-Thought

with tools, and memory-augmented agents. You build agents that recover from errors,

avoid infinite loops, and produce verifiable results.

Part 1: The ReAct Pattern (Reason + Act)

ReAct is the foundational pattern for all tool-using agents.

Thought: [Reason about what to do next]
Action: [Tool name and inputs]
Observation: [Result from tool]
... (repeat)
Final Answer: [Synthesised conclusion]

ReAct Prompt Template

REACT_SYSTEM_PROMPT = """You are an autonomous research agent. You have access to tools
to answer questions. Use them step by step.

For each step, follow this exact format:
Thought: [Your reasoning about what to do next]
Action: [Tool name]
Action Input: [JSON input to the tool]
Observation: [You will receive the tool result here]

When you have a complete answer, use:
Thought: I now have enough information to answer.
Final Answer: [Your complete, evidence-based answer]

Rules:
- Never guess — always verify with a tool
- If a tool fails, try an alternative approach
- Stop after 10 steps maximum
- If you cannot find the answer after 10 steps, say so clearly
"""

def build_react_prompt(question: str, tool_descriptions: list[dict]) -> str:
    tools_str = "\n".join([
        f"- {t['name']}: {t['description']}" for t in tool_descriptions
    ])
    return f"""{REACT_SYSTEM_PROMPT}

Available Tools:
{tools_str}

Question: {question}

Begin:
Thought:"""

Part 2: Tool Definition Framework

from typing import Callable, Any
from dataclasses import dataclass, field

@dataclass
class Tool:
    """A callable tool for an agent."""
    name: str
    description: str
    fn: Callable
    parameters: dict  # JSON Schema
    required_params: list[str] = field(default_factory=list)

    def run(self, **kwargs) -> str:
        """Execute the tool and return a string result."""
        try:
            result = self.fn(**kwargs)
            return str(result)
        except Exception as e:
            return f"Tool error: {type(e).__name__}: {e}"

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters
        }


class ToolRegistry:
    """Registry of available tools for an agent."""

    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool | None:
        return self._tools.get(name)

    def list_tools(self) -> list[dict]:
        return [t.to_dict() for t in self._tools.values()]

    def execute(self, name: str, inputs: dict) -> str:
        tool = self.get(name)
        if not tool:
            return f"Unknown tool: '{name}'. Available: {list(self._tools.keys())}"
        return tool.run(**inputs)

Part 3: Agent Loop Implementation

import json
import re
import logging
from typing import Generator

logger = logging.getLogger(__name__)


class AgentLoop:
    """
    A complete ReAct agent loop with:
    - Tool execution
    - Error recovery
    - Step limiting
    - Memory/scratchpad
    """

    def __init__(self, llm_fn: Callable, tool_registry: ToolRegistry,
                  max_steps: int = 10, verbose: bool = True):
        self.llm = llm_fn
        self.tools = tool_registry
        self.max_steps = max_steps
        self.verbose = verbose
        self.scratchpad: list[dict] = []

    def run(self, question: str) -> str:
        """Run the agent loop to answer a question."""
        self.scratchpad = []

        prompt = build_react_prompt(question, self.tools.list_tools())
        current_prompt = prompt

        for step in range(self.max_steps):
            if self.verbose:
                logger.info(f"=== Step {step + 1} ===")

            # Get LLM response
            response = self.llm(current_prompt)

            # Check for final answer
            if "Final Answer:" in response:
                final = response.split("Final Answer:")[-1].strip()
                self._log_step("final", question, final)
                return final

            # Parse action
            action, action_input = self._parse_action(response)
            if not action:
                # Malformed response — prompt for correction
                current_prompt += response + "\nObservation: [Parse error — please use the format: Action: tool_name and Action Input: {\"param\": \"value\"}]\nThought:"
                continue

            # Execute tool
            observation = self.tools.execute(action, action_input)

            if self.verbose:
                logger.info(f"Action: {action}({action_input}) → {observation[:200]}")

            self._log_step(action, action_input, observation)

            # Append to prompt
            current_prompt += (
                f"{response}\n"
                f"Observation: {observation}\n"
                f"Thought:"
            )

        return "Maximum steps reached. Could not complete the task."

    def _parse_action(self, text: str) -> tuple[str | None, dict]:
        """Parse Action and Action Input from LLM response."""
        action_match = re.search(r"Action:\s*(.+?)(?:\n|$)", text)
        input_match = re.search(r"Action Input:\s*({.+?})", text, re.DOTALL)

        if not action_match:
            return None, {}

        action = action_match.group(1).strip()

        if input_match:
            try:
                action_input = json.loads(input_match.group(1))
            except json.JSONDecodeError:
                action_input = {}
        else:
            action_input = {}

        return action, action_input

    def _log_step(self, action: str, inputs: Any, result: str) -> None:
        self.scratchpad.append({
            "action": action,
            "inputs": inputs,
            "result": result
        })

Part 4: Plan-and-Execute Pattern

For complex tasks that need a high-level plan before execution.

PLANNER_PROMPT = """You are a planning agent. Given a task, break it down into
a numbered list of concrete, executable steps. Each step should be specific enough
for an executor agent to complete it using available tools.

Task: {task}

Plan (numbered list of steps, max 8):"""

EXECUTOR_PROMPT = """Complete the following task step as part of a larger plan.

Overall goal: {goal}
Current step: {step}
Context from previous steps: {context}

Use the available tools to complete this specific step.
Then provide: Step Result: [your result]"""


class PlanAndExecuteAgent:
    """Two-phase agent: plan then execute each step."""

    def __init__(self, llm_fn: Callable, tool_registry: ToolRegistry):
        self.llm = llm_fn
        self.tools = tool_registry
        self.executor = AgentLoop(llm_fn, tool_registry, max_steps=5)

    def run(self, task: str) -> dict:
        # Phase 1: Plan
        plan_response = self.llm(PLANNER_PROMPT.format(task=task))
        steps = self._parse_plan(plan_response)

        results = []
        context = ""

        # Phase 2: Execute each step
        for i, step in enumerate(steps):
            logger.info(f"Executing step {i+1}: {step}")
            prompt = EXECUTOR_PROMPT.format(
                goal=task, step=step, context=context[-2000:]
            )
            result = self.executor.run(prompt)
            results.append({"step": step, "result": result})
            context += f"\nStep {i+1}: {step}\nResult: {result}\n"

        # Synthesise
        synthesis = self.llm(f"""
Given these results from executing a plan, provide a final answer.

Task: {task}

Results:
{context}

Final Answer:""")

        return {
            "task": task,
            "plan": steps,
            "step_results": results,
            "final_answer": synthesis
        }

    def _parse_plan(self, plan_text: str) -> list[str]:
        lines = plan_text.strip().split("\n")
        steps = []
        for line in lines:
            match = re.match(r"^\d+[\.\)]\s*(.+)", line.strip())
            if match:
                steps.append(match.group(1).strip())
        return steps

Part 5: Memory Patterns

from collections import deque

class AgentMemory:
    """Short and long-term memory for agents."""

    def __init__(self, short_term_size: int = 10):
        self.short_term = deque(maxlen=short_term_size)
        self.long_term: dict[str, str] = {}  # Key-value store
        self.conversation: list[dict] = []

    def remember(self, key: str, value: str) -> None:
        """Store in long-term memory."""
        self.long_term[key] = value

    def recall(self, key: str) -> str | None:
        """Retrieve from long-term memory."""
        return self.long_term.get(key)

    def add_to_conversation(self, role: str, content: str) -> None:
        self.conversation.append({"role": role, "content": content})
        self.short_term.append({"role": role, "content": content})

    def get_context_window(self, max_chars: int = 4000) -> str:
        """Get recent conversation context within character limit."""
        context = ""
        for turn in reversed(list(self.short_term)):
            line = f"{turn['role'].upper()}: {turn['content']}\n"
            if len(context) + len(line) > max_chars:
                break
            context = line + context
        return context

Part 6: Error Recovery Patterns

def with_retry(fn: Callable, max_retries: int = 3,
               backoff: float = 1.0) -> Callable:
    """Decorator for retrying failed agent steps."""
    def wrapper(*args, **kwargs):
        for attempt in range(max_retries):
            try:
                return fn(*args, **kwargs)
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                wait = backoff * (2 ** attempt)
                logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait}s...")
                time.sleep(wait)
    return wrapper


def fallback(primary_fn: Callable, fallback_fn: Callable) -> Callable:
    """Run primary, fall back to secondary on failure."""
    def wrapper(*args, **kwargs):
        try:
            return primary_fn(*args, **kwargs)
        except Exception as e:
            logger.warning(f"Primary failed ({e}), trying fallback...")
            return fallback_fn(*args, **kwargs)
    return wrapper

Output Standards

• Always set max_steps to prevent infinite loops (default: 10)
• Log every tool call with inputs and abbreviated output
• Include a scratchpad/reasoning trace in all agent outputs
• Use Plan-and-Execute for tasks with > 3 sequential dependencies
• Every tool must return a string (agents communicate via text)
• Test agents with adversarial inputs: empty results, tool failures, ambiguous questions

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)