How AI Engineers Build Reliable Agentic Systems

Agentic AI is no longer just a demo; it is becoming part of production systems now. But when agents start calling tools, manipulating state, and making decisions, one‑off prompts quickly break. The difference between a toy agent and a reliable system is architecture: planning, tools, memory, error handling, and evaluation built into the design.

In this post, you’ll see how an ML engineer can build a simple but realistic agentic system in Python, using:

An LLM as the reasoning engine.
Tools for external actions.
A minimal memory layer.
Retry and fallback logic.
Evaluatable traces.

This pattern maps to real‑world practices from industry coverage on reliable agentic systems, which emphasize modular, API‑first, and observable workflows.

1. A high‑level picture

A reliable agentic system usually looks like this:

User query enters the system.
Planner (LLM) breaks it into steps.
Tools execute concrete actions.
Memory stores context and prior results.
Verifiers check outputs and decide when to retry or stop.
Tracing logs everything so you can debug and test.

The goal is not to make the model perfect. It is to build a system that is robust even when the model makes mistakes.

2. Very simple agent core

We’ll code a minimal agent loop that:

Uses a model (here, a mock LLM class so you can plug in Ollama, OpenAI, etc.).
Maintains a state dict (state) as memory.
Calls a few simple tools.
Logs each step for later inspection.

Example tools

python
from typing import Dict, Any, List

class Tool:
    name: str
    description: str

    def run(self, state: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        raise NotImplementedError


class SearchTool(Tool):
    name = "search"
    description = "Search the web or internal docs for information."

    def run(self, state: Dict[str, Any], *, query: str) -> Dict[str, Any]:
        # In practice this would call a real search API or RAG.
        # Here, we mock an answer.
        return {
            "query": query,
            "result": f"Mock search result for: {query}",
        }


class CalculatorTool(Tool):
    name = "calculator"
    description = "Compute a numeric expression."

    def run(self, state: Dict[str, Any], *, expr: str) -> Dict[str, Any]:
        # In production, validate expr and avoid `eval`.
        try:
            # This is a demo; you should use a proper math library.
            value = eval(expr, {"__builtins__": {}}, {})
            if isinstance(value, (int, float)):
                return {"value": value, "success": True}
            else:
                return {"error": "Result not numeric", "success": False}
        except Exception as e:
            return {"error": str(e), "success": False}


TOOLS: Dict[str, Tool] = {
    "search": SearchTool(),
    "calculator": CalculatorTool(),
}

Minimal LLM “mock”

You’d replace this with any real LLM interface (e.g., OpenAI, Ollama, or a LangChain chain):

python
import json
from typing import List

class MockLLM:
    def __init__(self):
        self.history: List[str] = []

    def call(self, prompt: str) -> str:
        self.history.append(prompt)
        # In a real agent:
        #   response = client.chat.completions.create(...).choices.message.content
        # For this demo, assume a fixed decision pattern.
        if "search" in prompt.lower():
            return json.dumps({
                "action": "search",
                "arguments": {
                    "query": "What is the population of Earth?",
                },
            })
        if "calculator" in prompt.lower():
            return json.dumps({
                "action": "calculator",
                "arguments": {
                    "expr": "10 ** 2",
                },
            })
        if "all done" in prompt.lower():
            return "I have finished the task and the answer is 100."

        # Simple fallback response
        return "I don't know; please provide more context."

3. Agent loop with retries and state

Now we build an agent that:

Maintains state (memory).
Keeps a trace of steps.
Can retry failed steps.
Stops when the task is done.

python
from typing import Dict, Any, Optional
from datetime import datetime

MAX_STEPS = 10
MAX_RETRIES = 3


class AgenticSystem:
    def __init__(self):
        self.llm = MockLLM()
        self.state: Dict[str, Any] = {}
        self.trace: List[Dict[str, Any]] = []

    def plan_step(self, question: str) -> Optional[Dict[str, Any]]:
        # In a real system this would be a structured prompt
        # asking for {"action": "...", "arguments": { ... }, "reasoning": "..."}
        prompt = f"""
You are helping answer the user question:

Question: {question}

Current state: {self.state}

Decide one concrete action to take next. Choose from:

- search: to look up information.
- calculator: to compute a numeric expression.
- answer: to return a final answer.

Return your decision in JSON format:
{"action": "...", "arguments": { ... }}
        """.strip()

        try:
            raw = self.llm.call(prompt)
            # In a real app, use a stricter JSON parser and schema validation.
            return json.loads(raw)
        except Exception as e:
            self.log("plan_error", error=str(e))
            return None

    def run_tool(
        self, action: str, arguments: Dict[str, Any]
    ) -> Dict[str, Any]:
        result = {
            "action": action,
            "arguments": arguments,
            "success": False,
        }

        if action not in TOOLS:
            result["error"] = f"Unknown tool: {action}"
            return result

        tool = TOOLS[action]
        out = tool.run(self.state, **arguments)

        # Standardize tool response shape
        result.update(out)
        if "success" not in result:
            result["success"] = True

        return result

    def should_retry(self, tool_result: Dict[str, Any], step: int) -> bool:
        # Example policy: retry failed tools up to MAX_RETRIES, but not when already at MAX_STEPS.
        if not tool_result["success"] and step < MAX_STEPS:
            attempts = self.state.get("tool_attempts", {}).get(tool_result.get("action"), 0)
            return attempts < MAX_RETRIES
        return False

    def log(
        self,
        event: str,
        **extra: Any,
    ):
        self.trace.append(
            {
                "event": event,
                "step": len(self.trace),
                "timestamp": datetime.utcnow().isoformat(),
                **extra,
            }
        )

    def run(self, question: str, max_steps: int = MAX_STEPS) -> Dict[str, Any]:
        self.state = {"question": question, "tool_attempts": {}, "steps_taken": 0}
        self.trace = []

        step = 0
        final_answer = None

        while step < max_steps and final_answer is None:
            step += 1
            self.state["steps_taken"] = step

            self.log("step_start", step=step)

            # 1. Plan next step
            decision = self.plan_step(question)
            if decision is None:
                self.log("plan_failed", step=step)
                break

            action = decision.get("action")
            arguments = decision.get("arguments", {})
            self.log("planned_action", action=action, arguments=arguments)

            if action == "answer":
                # Assume the LLM intends to return a final answer.
                # In reality, you might call the LLM again with a fixed answer prompt.
                final_answer = arguments.get("text", "No answer provided.")
                self.log("answer_final", answer=final_answer)
                break

            # 2. Run tool
            tool_result = self.run_tool(action, arguments)

            # 3. Update state and trace
            self.log("tool_result", tool_result=tool_result)

            if not tool_result["success"]:
                self.log("tool_failed", action=action, step=step)

                # 4. Decide whether to retry
                if self.should_retry(tool_result, step):
                    self.state.setdefault("tool_attempts", {})
                    self.state["tool_attempts"][action] = self.state["tool_attempts"].get(action, 0) + 1
                    continue  # Retry this step (conceptually; in practice you might tell the LLM to retry)
                else:
                    final_answer = "I encountered an error and cannot complete the task."
                    self.log("task_failed", reason="too many failures")
                    break

            # 5. Update state with tool result if needed
            self.state["last_tool_result"] = tool_result

        return {
            "question": question,
            "final_answer": final_answer,
            "state": self.state,
            "trace": self.trace,
        }

4. Using the agent

python
agent = AgenticSystem()

result = agent.run(
    question="What is 10 squared?"
)

print("Final answer:", result["final_answer"])
print("Trace:")
for event in result["trace"]:
    print(f"  {event['event']}: {event.get('action', '')} → {event.get('tool_result', {}).get('value', '')}")

In practice, you’d:

plug in a real LLM client.
integrate retrieval (e.g., Qdrant or a BM25‑based index).
store traces in a database.
add metrics and alerts.

5. What makes this “reliable”

This simple example already embodies several patterns from current best‑practice guides for agentic systems:

Modular tooling: each tool is a small, testable unit.
Explicit state: state is a clear, inspectable object instead of buried in prompts.
Retry and fallback: the system can retry failed tools instead of crashing.
Trace‑driven evaluation: every step is logged so you can replay, debug, and test.

Reliability does not come from a single component. It comes from designing the whole system so that failures are expected, isolated, and recoverable.

6. From prototype to production

As you move this pattern to production, common next steps are:

Add retrieval (RAG) for grounding the agent in your data.
Add verification layers (self‑check, secondary models, rules).
Build an evaluation harness that tests multi‑step task success, not just single‑turn answers.
Expose observability (traces, metrics, dashboards) so ML engineers can track health.

For ML engineers, the core insight is simple: treat the agent as a system, not a prompt. Once you do that, the same engineering discipline you use for APIs, databases, and services transfers to agentic AI.

How AI Engineers Build Reliable Agentic Systems

In this post, you’ll see how an ML engineer can build a simple but realistic agentic system in Python, using:

An LLM as the reasoning engine.
Tools for external actions.
A minimal memory layer.
Retry and fallback logic.
Evaluatable traces.

This pattern maps to real‑world practices from industry coverage on reliable agentic systems, which emphasize modular, API‑first, and observable workflows.

1. A high‑level picture

A reliable agentic system usually looks like this:

User query enters the system.
Planner (LLM) breaks it into steps.
Tools execute concrete actions.
Memory stores context and prior results.
Verifiers check outputs and decide when to retry or stop.
Tracing logs everything so you can debug and test.

The goal is not to make the model perfect. It is to build a system that is robust even when the model makes mistakes.

2. Very simple agent core

We’ll code a minimal agent loop that:

Uses a model (here, a mock LLM class so you can plug in Ollama, OpenAI, etc.).
Maintains a state dict (state) as memory.
Calls a few simple tools.
Logs each step for later inspection.

Example tools

python
from typing import Dict, Any, List

class Tool:
    name: str
    description: str

    def run(self, state: Dict[str, Any], **kwargs) -> Dict[str, Any]:
        raise NotImplementedError


class SearchTool(Tool):
    name = "search"
    description = "Search the web or internal docs for information."

    def run(self, state: Dict[str, Any], *, query: str) -> Dict[str, Any]:
        # In practice this would call a real search API or RAG.
        # Here, we mock an answer.
        return {
            "query": query,
            "result": f"Mock search result for: {query}",
        }


class CalculatorTool(Tool):
    name = "calculator"
    description = "Compute a numeric expression."

    def run(self, state: Dict[str, Any], *, expr: str) -> Dict[str, Any]:
        # In production, validate expr and avoid `eval`.
        try:
            # This is a demo; you should use a proper math library.
            value = eval(expr, {"__builtins__": {}}, {})
            if isinstance(value, (int, float)):
                return {"value": value, "success": True}
            else:
                return {"error": "Result not numeric", "success": False}
        except Exception as e:
            return {"error": str(e), "success": False}


TOOLS: Dict[str, Tool] = {
    "search": SearchTool(),
    "calculator": CalculatorTool(),
}

Minimal LLM “mock”

You’d replace this with any real LLM interface (e.g., OpenAI, Ollama, or a LangChain chain):

python
import json
from typing import List

class MockLLM:
    def __init__(self):
        self.history: List[str] = []

    def call(self, prompt: str) -> str:
        self.history.append(prompt)
        # In a real agent:
        #   response = client.chat.completions.create(...).choices.message.content
        # For this demo, assume a fixed decision pattern.
        if "search" in prompt.lower():
            return json.dumps({
                "action": "search",
                "arguments": {
                    "query": "What is the population of Earth?",
                },
            })
        if "calculator" in prompt.lower():
            return json.dumps({
                "action": "calculator",
                "arguments": {
                    "expr": "10 ** 2",
                },
            })
        if "all done" in prompt.lower():
            return "I have finished the task and the answer is 100."

        # Simple fallback response
        return "I don't know; please provide more context."

3. Agent loop with retries and state

Now we build an agent that:

Maintains state (memory).
Keeps a trace of steps.
Can retry failed steps.
Stops when the task is done.

python
from typing import Dict, Any, Optional
from datetime import datetime

MAX_STEPS = 10
MAX_RETRIES = 3


class AgenticSystem:
    def __init__(self):
        self.llm = MockLLM()
        self.state: Dict[str, Any] = {}
        self.trace: List[Dict[str, Any]] = []

    def plan_step(self, question: str) -> Optional[Dict[str, Any]]:
        # In a real system this would be a structured prompt
        # asking for {"action": "...", "arguments": { ... }, "reasoning": "..."}
        prompt = f"""
You are helping answer the user question:

Question: {question}

Current state: {self.state}

Decide one concrete action to take next. Choose from:

- search: to look up information.
- calculator: to compute a numeric expression.
- answer: to return a final answer.

Return your decision in JSON format:
{"action": "...", "arguments": { ... }}
        """.strip()

        try:
            raw = self.llm.call(prompt)
            # In a real app, use a stricter JSON parser and schema validation.
            return json.loads(raw)
        except Exception as e:
            self.log("plan_error", error=str(e))
            return None

    def run_tool(
        self, action: str, arguments: Dict[str, Any]
    ) -> Dict[str, Any]:
        result = {
            "action": action,
            "arguments": arguments,
            "success": False,
        }

        if action not in TOOLS:
            result["error"] = f"Unknown tool: {action}"
            return result

        tool = TOOLS[action]
        out = tool.run(self.state, **arguments)

        # Standardize tool response shape
        result.update(out)
        if "success" not in result:
            result["success"] = True

        return result

    def should_retry(self, tool_result: Dict[str, Any], step: int) -> bool:
        # Example policy: retry failed tools up to MAX_RETRIES, but not when already at MAX_STEPS.
        if not tool_result["success"] and step < MAX_STEPS:
            attempts = self.state.get("tool_attempts", {}).get(tool_result.get("action"), 0)
            return attempts < MAX_RETRIES
        return False

    def log(
        self,
        event: str,
        **extra: Any,
    ):
        self.trace.append(
            {
                "event": event,
                "step": len(self.trace),
                "timestamp": datetime.utcnow().isoformat(),
                **extra,
            }
        )

    def run(self, question: str, max_steps: int = MAX_STEPS) -> Dict[str, Any]:
        self.state = {"question": question, "tool_attempts": {}, "steps_taken": 0}
        self.trace = []

        step = 0
        final_answer = None

        while step < max_steps and final_answer is None:
            step += 1
            self.state["steps_taken"] = step

            self.log("step_start", step=step)

            # 1. Plan next step
            decision = self.plan_step(question)
            if decision is None:
                self.log("plan_failed", step=step)
                break

            action = decision.get("action")
            arguments = decision.get("arguments", {})
            self.log("planned_action", action=action, arguments=arguments)

            if action == "answer":
                # Assume the LLM intends to return a final answer.
                # In reality, you might call the LLM again with a fixed answer prompt.
                final_answer = arguments.get("text", "No answer provided.")
                self.log("answer_final", answer=final_answer)
                break

            # 2. Run tool
            tool_result = self.run_tool(action, arguments)

            # 3. Update state and trace
            self.log("tool_result", tool_result=tool_result)

            if not tool_result["success"]:
                self.log("tool_failed", action=action, step=step)

                # 4. Decide whether to retry
                if self.should_retry(tool_result, step):
                    self.state.setdefault("tool_attempts", {})
                    self.state["tool_attempts"][action] = self.state["tool_attempts"].get(action, 0) + 1
                    continue  # Retry this step (conceptually; in practice you might tell the LLM to retry)
                else:
                    final_answer = "I encountered an error and cannot complete the task."
                    self.log("task_failed", reason="too many failures")
                    break

            # 5. Update state with tool result if needed
            self.state["last_tool_result"] = tool_result

        return {
            "question": question,
            "final_answer": final_answer,
            "state": self.state,
            "trace": self.trace,
        }

4. Using the agent

python
agent = AgenticSystem()

result = agent.run(
    question="What is 10 squared?"
)

print("Final answer:", result["final_answer"])
print("Trace:")
for event in result["trace"]:
    print(f"  {event['event']}: {event.get('action', '')} → {event.get('tool_result', {}).get('value', '')}")

In practice, you’d:

plug in a real LLM client.
integrate retrieval (e.g., Qdrant or a BM25‑based index).
store traces in a database.
add metrics and alerts.

5. What makes this “reliable”

This simple example already embodies several patterns from current best‑practice guides for agentic systems:

Modular tooling: each tool is a small, testable unit.
Explicit state: state is a clear, inspectable object instead of buried in prompts.
Retry and fallback: the system can retry failed tools instead of crashing.
Trace‑driven evaluation: every step is logged so you can replay, debug, and test.

Reliability does not come from a single component. It comes from designing the whole system so that failures are expected, isolated, and recoverable.

6. From prototype to production

As you move this pattern to production, common next steps are:

Add retrieval (RAG) for grounding the agent in your data.
Add verification layers (self‑check, secondary models, rules).
Build an evaluation harness that tests multi‑step task success, not just single‑turn answers.
Expose observability (traces, metrics, dashboards) so ML engineers can track health.

How AI Engineers Build Reliable Agentic Systems

In this article

How AI Engineers Build Reliable Agentic Systems

1. A high‑level picture

2. Very simple agent core

Example tools

Minimal LLM “mock”

3. Agent loop with retries and state

4. Using the agent

5. What makes this “reliable”

6. From prototype to production

ACTION_REQUIRED

Topics covered

BJ

// Related Articles

RAG with Kafka Introduction

Real-time Vector Ingestion via Kafka

The Production Gap: Why AI Prototypes Fail at Scale

How AI Engineers Build Reliable Agentic Systems

In this article

How AI Engineers Build Reliable Agentic Systems

1. A high‑level picture

2. Very simple agent core

Example tools

Minimal LLM “mock”

3. Agent loop with retries and state

4. Using the agent

5. What makes this “reliable”

6. From prototype to production

ACTION_REQUIRED

Topics covered

BJ

// Related Articles

RAG with Kafka Introduction

Real-time Vector Ingestion via Kafka

The Production Gap: Why AI Prototypes Fail at Scale

Let’s Architect Your Next System.

Let’s Architect Your Next System.

How AI Engineers Build Reliable Agentic Systems

In this article

How AI Engineers Build Reliable Agentic Systems

1. A high‑level picture

2. Very simple agent core

Example tools

Minimal LLM “mock”

3. Agent loop with retries and state

4. Using the agent

5. What makes this “reliable”

6. From prototype to production

ACTION_REQUIRED

Topics covered

BJ

// Related Articles

RAG with Kafka Introduction

Real-time Vector Ingestion via Kafka

The Production Gap: Why AI Prototypes Fail at Scale

Never Miss AI Breakthrough

How AI Engineers Build Reliable Agentic Systems

In this article

How AI Engineers Build Reliable Agentic Systems

1. A high‑level picture

2. Very simple agent core

Example tools

Minimal LLM “mock”

3. Agent loop with retries and state

4. Using the agent

5. What makes this “reliable”

6. From prototype to production

ACTION_REQUIRED

Topics covered

BJ

// Related Articles

RAG with Kafka Introduction

Real-time Vector Ingestion via Kafka

The Production Gap: Why AI Prototypes Fail at Scale

Never Miss AI Breakthrough

Let’s Architect Your Next System.

Let’s Architect Your Next System.