Ai Agents

AI Agents

Purpose

Designs production-grade AI agent systems with formally defined architectures, tool interfaces, memory hierarchies, multi-agent orchestration topologies, observability instrumentation, and safety guardrails.

Agent Protocol

Trigger

User request includes: AI agent, agentic system, tool-use, function calling, LangChain agent, CrewAI, AutoGen, multi-agent, agent memory, ReAct, Plan-and-Execute, agent orchestration, agent observability, agent safety, agent handoff, agent evaluation.

Input Context

• Task scope: single-turn vs. multi-turn vs. continuous operation
• Available tools and their APIs (REST, SDK, DB, shell)
• Number of agents and their roles
• Latency requirements (real-time vs. async)
• Token budget and cost constraints
• Existing infrastructure (LLM provider, vector DB, message queue)
• Security requirements (human-in-the-loop, data isolation)

Output Artifact

Agent system design document specifying architecture pattern, tool schemas, memory topology, orchestration protocol, error handling, and observability configuration.

Response Format

## Agent System Design

### Architecture Decision
Problem: {task description}
Selected Pattern: {ReAct | Plan-and-Execute | Reflection | Tool-Use | Hybrid}
Rationale: {why this pattern fits}
LLM: {model} | Provider: {provider} | Max Iterations: {N}
Temperature: {T} | Stop Conditions: {conditions}

### Tool Definitions
{tool_name}
  Description: {when to use / when NOT to use}
  Schema: {OpenAI / Anthropic tool format}
  Parameters: {name, type, required, description, enum}
  Returns: {type, structure}
  Safety: {validation rules, rate limit, idempotency}
  Error Handling: {retry strategy, fallback}

### Memory Architecture
Ephemeral: {conversation window N turns}
Working: {summary frequency, compression strategy}
Long-Term: {entity extraction | vector store | hybrid}
Persistence Layer: {Redis | PostgreSQL | Pinecone | custom}
Retention Policy: {TTL, max tokens, archival strategy}

### Orchestration Topology
Topology: {single-agent | supervisor | sequential | hierarchical | debate | mesh}
Agent Roles: [{name: primary responsibility, tools: [...]}]
Communication: {direct call | message queue | event bus | shared memory}
Handoff Protocol: {condition -> target agent, data contract}
Synchronization: {lock-step | async | event-driven}
Termination: {max rounds | consensus | human approval | error threshold}

### Error Handling
Transient Failures: {exponential backoff | immediate retry | circuit breaker}
Tool Errors: {return error to agent | skip | retry with modified args}
Agent Loops: {max iterations N, semantic dedup detection, entropy monitoring}
Escalation: {human handoff condition, fallback response}
Degradation: {partial results | cached response | simplified workflow}

### Observability
Tracing: {OpenTelemetry | Langfuse | custom}
Metrics: {latency p50/p95/p99, token usage, tool call count, error rate, loop count}
Logging: {structured JSON, agent_id, trace_id, turn_id}
Alerting: {anomaly detection, budget threshold, degradation events}

### Safety & Guardrails
Tool Access Control: {RBAC | allowlist | parameter constraints}
Input Validation: {prompt injection detection | parameter sanitization}
Rate Limiting: {calls/min, tokens/min, concurrent sessions}
Human Oversight: {destructive actions, high-cost operations, escalation path}
Audit Trail: {full action log, immutable storage, retention period}

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output.

Completion Criteria

• [ ] Architecture pattern selected and justified via decision tree.
• [ ] Tool schemas fully specified with validation rules and error handling.
• [ ] Memory tier architecture defined with retention policy.
• [ ] Orchestration topology documented with handoff protocol and termination conditions.
• [ ] Error handling covers transient failures, tool errors, loops, and escalation.
• [ ] Observability instrumentation specified (tracing, metrics, logging).
• [ ] Safety guardrails defined (access control, input validation, rate limiting, audit).
• [ ] Termination conditions prevent infinite execution and runaway costs.

Architecture Decision Framework

Decision Tree: Pattern Selection

Task requires external tool calls?
├── No → Is output quality the primary concern?
│   ├── Yes → Is the task iterative in nature (code, writing, analysis)?
│   │   ├── Yes → REFLECTION
│   │   └── No → STANDARD LLM CALL (not an agent problem)
│   └── No → STANDARD LLM CALL (not an agent problem)
└── Yes → How many sequential tool dependencies?
    ├── 0-1 → Is reasoning trace required for audit/debug?
    │   ├── Yes → ReAct
    │   └── No → TOOL-USE (function calling without reasoning)
    ├── 2-5 → Are sub-steps independently executable?
    │   ├── Yes → REACT (implicit planning in reasoning loop)
    │   └── No → REACT with explicit sub-goal decomposition
    └── >5 → Does the task decompose into independent subtasks?
        ├── Yes → PLAN-AND-EXECUTE
        └── No → HYBRID (Planner → ReAct per subtask)

Decision Tree: Memory Tier Selection

Agent needs to retain info across sessions?
├── No → How many turns in single session?
│   ├── 50 → CONVERSATION + SUMMARY + VECTOR MEMORY
└── Yes → What type of cross-session recall?
    ├── Facts about user → ENTITY MEMORY (extract + store attributes)
    ├── Past conversation context → VECTOR MEMORY (semantic retrieval)
    ├── Both → HYBRID (entity + vector)
    └── Task progress → WORKING MEMORY (structured state machine)

Decision Tree: Multi-Agent Topology Selection

How many agents are needed?
├── 1 → SINGLE-AGENT (all tools to one agent)
├── 2-5 → Do agents perform different functions?
│   ├── No → DEBATE (same goal, different perspectives)
│   └── Yes → Do tasks have sequential dependency?
│       ├── Yes → SEQUENTIAL (pipeline)
│       └── No → SUPERVISOR (delegator pattern)
└── >5 → Do you need nested specialization?
    ├── Yes → HIERARCHICAL (supervisor of supervisors)
    └── No → MESH (peer-to-peer, any agent can route to any other)

Architectural Patterns

Pattern 1: ReAct (Reasoning + Acting)

The foundational agent pattern where the model interleaves reasoning (Thought) with action (Tool call) in a loop until a final answer is produced.

Core Loop: Observation → Thought → Action → Observation → ... → Final Answer

Implementation Strategy:

• System prompt defines the Thought/Action/Observation format
• Each iteration appends the full turn to the message list
• Action is parsed from structured output (JSON) or text format
• Tool result is injected as a tool-response message

When to Use:

• Multi-step tasks requiring external data retrieval
• Tasks where audit trail of reasoning is required
• Default pattern for most single-agent systems
• Situations where the model must adapt its plan based on intermediate results

When NOT to Use:

• Single tool call (use Tool-Use pattern, cheaper)
• Tasks with >10 sequential dependencies (use Plan-and-Execute)
• Latency-critical applications (each reasoning step adds 500-2000ms)

Anti-Patterns:

• Letting the agent loop without semantic termination detection
• Overly permissive tool access leads to hallucinated tool calls
• Missing structured output parsing causes fragile action extraction

Code Implementation:

class ReActAgent:
    def __init__(self, llm, tools, max_iterations=15, loop_detector=None):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_iterations = max_iterations
        self.loop_detector = loop_detector or SemanticLoopDetector(window=5)
        self.trace = []

    def run(self, task: str, context: dict = None) -> dict:
        messages = _build_system_messages(self.tools)
        messages.append({"role": "user", "content": task})
        if context:
            messages.append({"role": "system", "content": f"Context: {json.dumps(context)}"})

        for turn in range(self.max_iterations):
            response = self.llm.invoke(messages)
            messages.append(response)

            if response.stop_reason == "end_turn":
                return {"result": response.content, "turns": turn + 1, "trace": self.trace}

            action = self._parse_action(response)
            if action is None:
                continue

            self.trace.append({"turn": turn, "thought": action.thought, "tool": action.name, "args": action.args})

            if self.loop_detector.detect(self.trace):
                return {"result": "Loop detected. Returning best effort.", "turns": turn + 1, "trace": self.trace}

            if action.name in self.tools:
                try:
                    result = self.tools[action.name].execute(**action.args)
                    messages.append({"role": "tool", "content": truncate(str(result), 10000), "tool_call_id": action.call_id})
                except Exception as e:
                    messages.append({"role": "tool", "content": f"Error: {str(e)}", "tool_call_id": action.call_id})
            else:
                messages.append({"role": "tool", "content": f"Error: Tool '{action.name}' not found.", "tool_call_id": action.call_id})

        return {"result": "Max iterations reached.", "turns": self.max_iterations, "trace": self.trace}

    def _parse_action(self, response):
        if response.content and "Final Answer:" in response.content:
            return None  # Will be caught by stop_reason check in production
        try:
            return ToolCall.parse(response.tool_calls[0]) if response.tool_calls else None
        except (IndexError, KeyError):
            return None

Advanced ReAct Variants:

• ReAct with Reflection: After each action-result pair, the agent reflects on whether the result makes sense before proceeding
• ReAct with Verification: After the final answer, a separate verification step checks correctness
• Tree-of-ReAct: Multiple ReAct trajectories run in parallel, best path selected at the end

---

Pattern 2: Plan-and-Execute

Decomposes complex tasks into a plan of independent subtasks, executes each, then synthesizes results.

Flow: Task → Planner → [Subtask 1, Subtask 2, ..., Subtask N] → Executor Pool → Synthesizer → Final Output

Planner Design:

class Planner:
    def create_plan(self, task: str, max_subtasks: int = 10) -> Plan:
        prompt = f"""
        Break this task into sequential subtasks. Each subtask must be:
        1. Self-contained (executable independently)
        2. Observable (clear completion signal)
        3. Order-specified (dependencies declared)

        Task: {task}

        Output as JSON array:
        [{{"id": 1, "description": "...", "depends_on": [], "tools_needed": [...]}}, ...]
        """
        response = self.llm.invoke(prompt)
        return Plan.parse(response.content)

Re-Planning Strategy: When a subtask fails, the planner can either:

• Retry the same subtask with modified parameters
• Re-plan remaining subtasks accounting for the failure
• Abort and escalate if the failure is unrecoverable

Executor Pool Architecture:

class ExecutorPool:
    def __init__(self, max_concurrent=3):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_all(self, plan: Plan, context: dict) -> list:
        results = {}
        async def run_subtask(st):
            async with self.semaphore:
                result = await self._execute_single(st, context, results)
                results[st.id] = result
                return result

        # Topological execution respecting dependencies
        ready = [st for st in plan.subtasks if not st.depends_on]
        pending = {st.id: st for st in plan.subtasks if st.depends_on}

        while ready:
            batch = [run_subtask(st) for st in ready]
            completed = await asyncio.gather(*batch, return_exceptions=True)
            ready = []
            for st_id, st in list(pending.items()):
                if all(dep in results for dep in st.depends_on):
                    ready.append(st)
                    del pending[st_id]

        return results

Synthesizer Patterns:

• Concatenation: Simple ordered assembly (for linear workflows)
• Template Filling: Insert results into a structured template (for reports)
• Semantic Merge: LLM merges results into coherent output (for complex synthesis)
• Voting: Multiple approaches, best selected by quality metric

When to Use:

• Tasks with 5+ sequential dependencies
• Complex workflows where planning cost is justified
• Scenarios requiring intermediate verification before proceeding
• Parallelizable subtasks benefit from executor pool

When NOT to Use:

• Simple 1-3 step tasks (ReAct overhead is lower)
• Dynamic tasks where the plan must change based on results (use ReAct)
• Latency-critical applications (planning phase adds 2-5 seconds)

---

Pattern 3: Reflection

The agent generates output, critiques it against criteria, and revises iteratively.

Loop: Generate → Critique → Revise → ... → Score >= Threshold → Final

Critic Configurations:

• Same Model Self-Critique: Most common, no additional cost for separate model
• Different Model Critique: Stronger model reviews weaker model's output
• Rule-Based Critique: Check output against formal constraints (schema, length, format)
• Multi-Aspect Critique: Different critics for different quality dimensions

class ReflectionAgent:
    def __init__(self, generator, critic, max_rounds=3, threshold=0.8):
        self.generator = generator
        self.critic = critic
        self.max_rounds = max_rounds
        self.threshold = threshold

    def run(self, task: str, criteria: list[str]) -> dict:
        output = self.generator.generate(task)
        rounds = 0

        for i in range(self.max_rounds):
            score = self.critic.evaluate(output, criteria)
            if score >= self.threshold:
                return {"output": output, "rounds": i + 1, "final_score": score}

            critique = self.critic.critique(output, criteria)
            output = self.generator.revise(output, critique)
            rounds = i + 1

        final_score = self.critic.evaluate(output, criteria)
        return {"output": output, "rounds": rounds, "final_score": final_score, "threshold_not_met": final_score  list[Message]:
        context = []
        context.extend(self.ephemeral.get_recent())
        semantic = self.vector_store.search(embed(query), k=5)
        if semantic:
            context.append({"role": "system", "content": f"Relevant past context: {semantic}"})
        entities = self.entity_store.get_relevant(query)
        if entities:
            context.append({"role": "system", "content": f"Known entities: {entities}"})
        return context

Multi-Agent Orchestration

Topology Reference

| Topology | Agents | Communication | Coordination | Fault Tolerance | |----------|--------|---------------|--------------|-----------------| | Single-Agent | 1 | N/A | N/A | None | | Supervisor | N+1 | Direct call | Centralized | Single point of failure | | Sequential | N | Message passing | Chain | Break on any failure | | Hierarchical | N+M | Tree routing | Multi-level | Partial | | Debate | N | Shared context | Consensus | Majority voting | | Mesh | N | Peer-to-peer | Distributed | High (redundant paths) | | Blackboard | N | Shared workspace | Tuple-space | High (decoupled agents) |

Handoff Protocol Design

@dataclass
class AgentHandoff:
    source_agent: str
    target_agent: str
    context: dict  # Carry-forward data
    handoff_reason: str  # Why this handoff is happening
    priority: int  # 1-5, for queue ordering
    timeout_ms: int
    retry_policy: RetryPolicy

class HandoffProtocol:
    def __init__(self, registry: AgentRegistry):
        self.registry = registry

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [j4flmao](https://github.com/j4flmao)
- **Source:** [j4flmao/agent-skills](https://github.com/j4flmao/agent-skills)
- **License:** MIT

Install and usage instructions live in the source repository linked above.