DEV Community: MUHAMMAD MUSADIQ

How to Implement Strict JSON Schema Validation and Retry Logic for Agent Function Parameters

MUHAMMAD MUSADIQ — Thu, 30 Jul 2026 05:05:41 +0000

Building AI agents is fun until your LLM decides to pass a string into an integer field or hallucinates a non-existent key inside a tool call. I spent three hours last week debugging a pipeline because the OpenAI API returned "age": "twenty-five" instead of an integer. When building production-ready agents, generic function calling isn't enough. You need strict schema validation and an automatic self-correction loop to catch malformed payloads before they crash your backend.

In this guide, I will show you how to combine Pydantic schemas with a dynamic retry loop. This pattern ensures your agents recover from invalid JSON outputs gracefully, dramatically improving agent reliability.

Most LLM providers offer function calling features. You give the model a JSON schema, and it attempts to return structured data matching that schema.

However, models still make mistakes. They might pass invalid emails, out-of-range numbers, or fail to populate required parameters when handling complex tasks. If your application attempts to parse this data directly, your backend throws an exception and the user task fails.

To fix this, we need three distinct layers:

Strict Schema Validation: Using Pydantic to enforce exact types, ranges, and patterns.
Runtime Error Capture: Catching validation failures instantly.
Feedback-Based Retry Logic: Feeding the exact Pydantic error message back to the LLM so it can correct its own output.

If you are building complex systems like workflow automation pipelines, this extra layer of safety is the difference between an agent that works in demo videos and one that runs reliably in production.

Step 1: Define Your Strict Pydantic Schema

We will start by defining the tool parameters using Pydantic V2. Instead of relying on basic JSON types, we can add field validation rules, regular expressions, and custom validators.

Here is an example schema for a financial transfer execution tool:

from pydantic import BaseModel, Field, field_validator
from typing import Literal

class TransferFundsSchema(BaseModel):
    account_id: str = Field(
        ..., 
        pattern=r"^ACC-\d{6}$", 
        description="The account ID, formatted like ACC-123456"
    )
    amount: float = Field(
        ..., 
        gt=0, 
        le=10000, 
        description="Transfer amount in USD. Must be greater than 0 and at most 10,000"
    )
    currency: Literal["USD", "EUR", "GBP"] = Field(
        default="USD", 
        description="3-letter currency code"
    )
    memo: str = Field(
        ..., 
        min_length=5, 
        max_length=100, 
        description="Reason for transfer"
    )

    @field_validator("memo")
    @classmethod
    def memo_must_not_contain_sql(cls, value: str) -> str:
        forbidden = ["DROP", "SELECT", "DELETE", "UPDATE"]
        if any(word in value.upper() for word in forbidden):
            raise ValueError("Memo contains forbidden SQL keywords.")
        return value

Look at how explicit this schema is. We enforce exact string patterns, strict monetary bounds, explicit currency options, and even basic sanitization. If an LLM sends "amount": -50 or an invalid account string like "12345", Pydantic immediately throws a ValidationError.

Step 2: Extract OpenAI Structured Outputs JSON Schema

Modern versions of OpenAI allow you to pass Pydantic models directly using response formats or tool definitions. But under the hood, Pydantic converts your model to a standard JSON Schema dictionary.

Here is how you extract the JSON schema to send to your model:

import json

def get_tool_definition():
    return {
        "type": "function",
        "function": {
            "name": "transfer_funds",
            "description": "Executes a secure financial transfer.",
            "parameters": TransferFundsSchema.model_json_schema()
        }
    }

This guarantees your LLM receives the exact structure defined by your Python codebase.

Step 3: Build the Self-Correcting Retry Wrapper

This is where the magic happens. When the LLM generates arguments for transfer_funds, we intercept the arguments, parse them using TransferFundsSchema.model_validate_json(), and check for errors.

If parsing succeeds, we execute the action. If it fails, we catch the ValidationError, format the human-readable error messages, and send them back to the LLM inside the tool message history.

import openai
from pydantic import ValidationError

client = openai.OpenAI()

def run_agent_with_retry(user_prompt: str, max_retries: int = 3):
    messages = [
        {"role": "system", "content": "You are a helpful assistant. Call tools accurately based on user requests."},
        {"role": "user", "content": user_prompt}
    ]

    tools = [get_tool_definition()]

    for attempt in range(max_retries):
        print(f"\n--- Attempt {attempt + 1} ---")

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        message = response.choices[0].message
        messages.append(message)

        # Check if the LLM requested a tool call
        if not message.tool_calls:
            print("Agent responded with text instead of a tool call:")
            print(message.content)
            return message.content

        tool_call = message.tool_calls[0]
        tool_name = tool_call.function.name
        raw_args = tool_call.function.arguments

        print(f"LLM called tool '{tool_name}' with raw args: {raw_args}")

        if tool_name == "transfer_funds":
            try:
                # Validate raw JSON against our Pydantic model
                validated_data = TransferFundsSchema.model_validate_json(raw_args)
                print("Validation Successful!")

                # Execute your real business logic here
                return execute_transfer(validated_data)

            except ValidationError as e:
                # Format validation errors clearly for the model
                error_details = e.errors(include_url=False)
                clean_errors = [
                    f"Field '{'->'.join(str(loc) for loc in err['loc'])}': {err['msg']}" 
                    for err in error_details
                ]
                error_message = "Schema validation failed:\n" + "\n".join(clean_errors)

                print(f"Validation Error caught: {error_message}")

                # Append the validation error as a tool response
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": f"ERROR: Your input arguments were invalid.\n{error_message}\nPlease fix the arguments and try calling the tool again."
                })

    raise RuntimeError("Agent failed to provide valid tool inputs within the retry limit.")

def execute_transfer(data: TransferFundsSchema):
    return f"Successfully transferred ${data.amount} {data.currency} to {data.account_id}."

Step 4: Test the Resilience of the Retry Logic

Let's test this system with an ambiguous prompt that forces the LLM to guess fields incorrectly on purpose.

if __name__ == "__main__":
    # The prompt deliberately provides bad data format to trigger validation failures
    prompt = "Transfer 15000 dollars to account 999111 for contract payment standard UPDATE query."

    try:
        result = run_agent_with_retry(prompt)
        print("\nFinal Result:", result)
    except Exception as err:
        print("\nExecution Failed:", err)

What Happens Behind the Scenes:

Attempt 1: The LLM sees "account 999111" and sends "account_id": "999111". It also passes "amount": 15000.
Pydantic Triggers Errors:
- account_id fails the regex ^ACC-\d{6}$.
- amount fails the maximum boundary check (le=10000).
- memo contains forbidden SQL word "UPDATE".
Error Serialization: The detailed error string goes straight back to the LLM message array.
Attempt 2: The LLM reads the errors, fixes the formatting to "ACC-999111", adjusts the amount down to $10,000, fixes the memo, and submits valid JSON.
Success: Parsing succeeds, and execution completes.

By combining strict schema validation with natural language feedback loops, you eliminate almost all runtime payload errors.

Next Steps for Production Systems

If you want to scale this pattern in production:

Limit Max Retries: Set retries to 2 or 3. If an agent fails 3 times in a row, fall back to human intervention.
Token Cost Optimization: Retries consume tokens. Ensure your validation error messages are direct and minimal.
Log Failures: Store repeated validation failures in your logging system. They usually reveal flaws in your prompt descriptions or schema design.

When building production-ready AI agent development workflows, handling real-world edge cases is essential. If your team needs help scaling fault-tolerant AI solutions or custom LLM infrastructure, Gaper provides experienced software engineers to help build resilient enterprise applications.

Why Custom-Built AI Agents Exceed Off-The-Shelf SaaS Limits in Complex Enterprise Systems

MUHAMMAD MUSADIQ — Thu, 30 Jul 2026 04:59:47 +0000

Last month, I watched an engineering director at a large logistics firm nearly throw his laptop through a window. His team had spent six months trying to force a popular off-the-shelf AI platform to work with their legacy inventory system. On paper, the software promised plug-and-play intelligence within days. In reality, it was like trying to plug a USB-C cable into a brick wall. Every custom database field broke the integration, API rate limits killed their nightly updates, and the vendor's closed-source model kept hallucinating nonexistent shipping rules.

This scenario plays out in enterprise tech stacks constantly. Everyone wants the shortcut. Executives see a polished demo from an AI startup and assume it will magically stitch together twenty years of technical debt. But when you are dealing with deeply embedded logic, proprietary data formats, and strict security boundaries, off-the-shelf SaaS hits a brick wall fast.

Off-the-shelf AI tools are great for general tasks. If you need a chatbot to summarize internal documentation or draft customer service emails, buying a subscription makes total sense. Do not rebuild the wheel for generic problems.

The trouble begins when you try to insert a generic tool into a complex workflow. Most SaaS AI vendors build their products to serve the widest possible market. To do that, they lock down the backend, hide the underlying prompt engineering, and expose only a few basic API endpoints. They give you a rigid box and tell you to fit your business inside it.

Real enterprise software does not fit in neat boxes. You have custom database schemas, obscure legacy APIs, dynamic authentication flows, and business rules refined over decades. When you try to push a generic AI agent to execute multi-step transactions across these systems, the abstraction layer crumbles. You end up spending more time engineering workarounds around the vendor's limitations than you would have spent building a custom solution from scratch.

Why Custom Workflows Change the Game

Building dedicated enterprise AI agents gives you complete control over state management, tool usage, and retrieval logic. Instead of hoping a third-party service handles your payload correctly, you build the orchestration layer yourself.

When you control the stack, you can design custom AI agents that talk directly to your internal microservices. You can give the agent exact specifications for function calling, enforce strict output schemas with validation libraries, and set up deterministic fallbacks when a model fails.

If a custom model gets confused by an edge case, you do not open a support ticket with a SaaS vendor and wait three weeks for a canned response. You inspect the execution trace, tweak your prompt routing or vector store retrieval, deploy a fix, and move on. You own the pipeline, the data context, and the feedback loops.

Specialized Domain Realities

This gap between generic tools and custom code gets even wider in regulated industries. Standard SaaS platforms usually operate on broad data models. They are not built for the nuance required in specialized fields.

Take financial auditing or risk assessment. A generic LLM wrapper might look impressive when parsing a basic invoice, but it fails when it has to reconcile disparate ledger systems across five international subsidiaries with different compliance rules. In domains like AI for accounting and finance, a hallucinated number or a skipped approval check is not just a minor bug, it is a massive audit failure.

Custom agents allow developers to build hard guardrails around the model. You can write code that forces the agent to cross-check every output against deterministic backend logic before any action is committed. You get the intelligence of an LLM paired with the predictability of traditional code. SaaS tools rarely offer that level of granular control.

The Reality of Build vs. Buy

Choosing to build does not mean writing your own foundational model from scratch. Nobody expects enterprises to train a trillion-parameter LLM in-house.

Building in 2026 means assembling specialized components. You take proven open or closed foundation models, connect them through custom orchestration frameworks, plug in your own vector databases, and wire them straight into your internal APIs.

It takes engineering effort up front. It requires actual software developers who understand prompt routing, context window management, and system architecture. But the payoff is an asset you actually own, tailored precisely to how your business operates, scaling without per-seat tax increases from a third-party vendor.

If your core workflows are simple and standard, buy off the shelf. But if your software architecture is your competitive advantage, trying to squeeze it into a generic AI SaaS box is a losing strategy. When off-the-shelf tools hit their limit, bringing in technical talent from platforms like Gaper can help teams build the exact, custom AI infrastructure they actually need.

How to Implement Dynamic Query Decomposition in Agentic RAG for Complex Enterprise Search

MUHAMMAD MUSADIQ — Thu, 30 Jul 2026 04:39:54 +0000

Standard RAG systems fall apart the moment a user asks a complex enterprise question. If someone asks, "What were our main operational risks in Q3, and how do they compare to our compliance updates in Q4?", a standard embedding model conflates those distinct concepts. The vector search then returns a messy compromise that misses critical context from both domains. I ran into this exact wall last year while building an internal search engine for cross-departmental documents, and it drove me crazy until we switched to dynamic query decomposition inside an agentic RAG pipeline.

When you deal with complex enterprise search, single-shot retrieval fails. You need an intelligent orchestration layer that breaks a complex prompt into small, focused sub-queries, executes targeted vector search calls for each piece, and synthesizes the retrieved contexts.

Here is how you can implement dynamic query decomposition step by step.

Standard vector search maps an entire user input into a single point in a high-dimensional vector space. When the input contains multiple distinct questions, conditional logic, or temporal comparisons, the resulting vector ends up in a semantic "middle ground." It loses the specific nuances of each sub-topic.

Agentic RAG solves this by introducing dynamic routing and query decomposition. Instead of sending the original user prompt straight to the vector database, an LLM retrieval agent inspects the prompt, identifies distinct information needs, generates targeted sub-queries, and routes them to the appropriate search indexes.

If you work on domain-specific systems such as AI for Accounting & Finance, this approach is mandatory. Financial audits and regulatory compliance questions almost always require pulling data across distinct quarters, departments, and policy documents simultaneously.

Step 1: Define the Structured Output Schemas

First, set up your Python environment. You need Pydantic to enforce strict structured outputs from your LLM decomposer.

from pydantic import BaseModel, Field
from typing import List

class SubQuery(BaseModel):
    id: int = Field(..., description="Unique index for the sub-query")
    query: str = Field(..., description="The specific, standalone query text to search for")
    target_collection: str = Field(..., description="The target vector index or collection, e.g., 'financial_reports', 'legal_docs', or 'ops_logs'")

class QueryDecompositionPlan(BaseModel):
    original_query: str = Field(..., description="The initial query provided by the user")
    sub_queries: List[SubQuery] = Field(..., description="List of decomposed sub-queries needed to answer the original prompt completely")
    reasoning: str = Field(..., description="Brief explanation of why the query was broken down this way")

Using Pydantic ensures your decomposer outputs valid JSON every single time, which prevents parse errors in production.

Step 2: Build the LLM Query Decomposer

Now write the core decomposition engine. This module takes a complex prompt and uses an LLM to split it into clean, isolated vector search queries.

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

DECOMPOSER_SYSTEM_PROMPT = """
You are an expert query planning agent for an enterprise retrieval system.
Your job is to break down complex, multi-part user prompts into concise, independent sub-queries.

Rules:
1. Each sub-query must be completely self-contained. Replace ambiguous pronouns with explicit nouns.
2. Identify the target document collection for each sub-query ('finance', 'legal', 'operations', or 'general').
3. Keep the number of sub-queries minimal, typically between 2 and 4.
4. If the user query is already simple and single-topic, return a list containing only the original query.
"""

def decompose_query(user_prompt: str) -> QueryDecompositionPlan:
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": DECOMPOSER_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt}
        ],
        response_format=QueryDecompositionPlan,
        temperature=0.0
    )
    return response.choices[0].message.parsed

# Example usage
prompt = "Compare our Q3 legal liabilities with the new Q4 compliance requirements for GDPR."
plan = decompose_query(prompt)

for sq in plan.sub_queries:
    print(f"[{sq.id}] Collection: {sq.target_collection} | Query: {sq.query}")

If you test this with the sample prompt, the model splits the request into two clean searches: one for Q3 legal liabilities in the legal store, and another for Q4 GDPR compliance rules.

Step 3: Run Parallel Vector Search Operations

Once you have your clean sub-queries, execute them concurrently. Running vector searches sequentially adds unnecessary latency to your application.

import asyncio
from typing import Dict, List

# Dummy representation of a vector store lookup
async def execute_vector_search(sub_query: SubQuery) -> Dict:
    # In practice, call your vector DB here (e.g., Qdrant, Pinecone, Weaviate)
    await asyncio.sleep(0.2)  # Simulate network latency

    fake_results = {
        "finance": "Q3 net income increased by 14% quarter-over-quarter.",
        "legal": "Q3 pending litigation costs were estimated at $1.2M.",
        "operations": "GDPR compliance framework updated in Q4 with mandatory 72-hour breach notifications."
    }

    retrieved_text = fake_results.get(sub_query.target_collection, "No matching records found.")

    return {
        "sub_query_id": sub_query.id,
        "query": sub_query.query,
        "collection": sub_query.target_collection,
        "retrieved_context": retrieved_text
    }

async def run_parallel_retrieval(plan: QueryDecompositionPlan) -> List[Dict]:
    tasks = [execute_vector_search(sq) for sq in plan.sub_queries]
    results = await asyncio.gather(*tasks)
    return results

This asynchronous approach scales nicely. Whether your agent generates two sub-queries or four, the total retrieval time remains roughly equal to a single query execution.

If your team is scaling these architectures across enterprise teams, setting up proper pipeline foundations matters early on. Dedicated engineering resources specializing in AI agent development can help optimize these parallel loops to prevent infrastructure bottlenecks.

Step 4: Synthesize Retrieved Contexts into the Final Answer

Now that you have contexts pulled from multiple vector search calls, feed them to a synthesizer LLM. The synthesizer blends the disparate pieces of context into one coherent response.

def synthesize_final_answer(original_query: str, retrieval_results: List[Dict]) -> str:
    formatted_contexts = ""
    for res in retrieval_results:
        formatted_contexts += f"\n--- Context from Query: '{res['query']}' (Index: {res['collection']}) ---\n"
        formatted_contexts += f"{res['retrieved_context']}\n"

    system_prompt = """
    You are an enterprise AI assistant. Answer the user's question using ONLY the provided retrieved contexts.
    If the context items contain distinct pieces of data, compare and synthesize them logically.
    Acknowledge any gaps if the information is incomplete.
    """

    user_content = f"Original Question: {original_query}\n\nRetrieved Contexts:\n{formatted_contexts}"

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_content}
        ],
        temperature=0.2
    )

    return response.choices[0].message.content

# Putting it together
async def main():
    user_prompt = "What were our Q3 legal liabilities and how do they impact our Q4 operational changes?"

    print("Decomposing query...")
    plan = decompose_query(user_prompt)

    print("Executing vector searches...")
    results = await run_parallel_retrieval(plan)

    print("Synthesizing final answer...")
    final_output = synthesize_final_answer(user_prompt, results)

    print("\nFinal Answer:\n", final_output)

# Run loop
asyncio.run(main())

Step 5: Add an Agentic Verification Loop

The real magic of agentic RAG happens when you add reflection. What if the vector search returns garbage for one of the sub-queries? A basic pipeline fails silently. An agentic system evaluates the retrieved context and dynamically retries with a modified search strategy.

Here is a simple evaluation node to handle missing data:

class RetrievalEvaluation(BaseModel):
    is_sufficient: bool = Field(..., description="True if retrieved contexts contain enough info to answer the prompt")
    missing_information: str = Field("", description="Description of missing details if context is insufficient")
    suggested_rewrites: List[str] = Field(default_factory=list, description="Alternative queries to try if search failed")

def evaluate_retrieval(original_query: str, retrieval_results: List[Dict]) -> RetrievalEvaluation:
    eval_prompt = f"""
    Evaluate if the following retrieved contexts contain enough information to answer this prompt:
    Prompt: {original_query}

    Contexts: {retrieval_results}
    """

    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[{"role": "user", "content": eval_prompt}],
        response_format=RetrievalEvaluation,
        temperature=0.0
    )
    return response.choices[0].message.parsed

If is_sufficient returns False, your orchestrator loop sends suggested_rewrites back to the vector search layer before presenting an answer to the user. This simple feedback loop turns static RAG into a resilient software system.

Wrapping Up

Building enterprise search that actually works requires moving past simple single-prompt embeddings. Dynamic query decomposition breaks complex prompts into manageable tasks, executes vector searches in parallel, and merges contexts cleanly.

Building custom agentic pipelines at scale often requires dedicated engineering expertise to handle production edge cases, latency budgets, and specialized domain integrations. If your organization needs help bringing production-ready AI systems to life, platforms like Gaper connect you with vetted talent to accelerate your technical roadmap.

How to Implement Context-Preserving Human Handoff Protocols for AI Support Agents

MUHAMMAD MUSADIQ — Thu, 30 Jul 2026 04:32:47 +0000

Customers hate repeating themselves when an AI bot fails and transfers them to a human agent. If your AI bot loses conversation history during a handoff, your customer satisfaction scores will plummet, and your support team will waste time asking questions the customer has already answered. You can solve this by building a clean handoff protocol that serializes the conversation state, generates an executive summary, and posts it directly to your ticketing system before routing the customer.

Here is what you need to follow this guide:

A running AI support agent built on Node.js or Python. If you are starting from scratch, check out this guide on AI agent development to structure your architecture.
Admin access to Zendesk with API token generation enabled.
A vector store or session cache, like Redis, holding current user conversation state.
Basic understanding of REST APIs and Webhooks.

Do not dump raw chat logs into a support ticket and expect your team to read fifty lines of back and forth. Use your LLM to summarize the key points right when the agent decides to trigger a handoff.

Define a simple schema for the summary payload:

{
  "customer_intent": "Refund Request",
  "resolved_points": ["Verified account ownership", "Located order #8841"],
  "unresolved_issue": "System rejected automated refund due to policy threshold",
  "user_sentiment": "Frustrated",
  "recommended_action": "Manually override refund block in billing portal"
}

Run a fast, cheap model like GPT-4o-mini with a dedicated prompt to extract these fields from the raw chat memory before calling the Zendesk API.

Step 2: Format the Internal Note for Zendesk

Zendesk allows you to attach private notes to tickets that are visible only to agents, not customers. Format your AI summary as a Markdown string so it displays cleanly in the Zendesk Agent Workspace.

Here is a simple Node.js helper function to construct the ticket payload:

async function createZendeskHandoffTicket(user, summary, transcript) {
  const ticketData = {
    ticket: {
      subject: `AI Handoff: ${summary.customer_intent} - ${user.name}`,
      comment: {
        body: `--- AI HANDOFF SUMMARY ---
Intent: ${summary.customer_intent}
Sentiment: ${summary.user_sentiment}

Key Information Gathered:
${summary.resolved_points.map(p => `- ${p}`).join('\n')}

Blocking Issue:
${summary.unresolved_issue}

Suggested Next Step:
${summary.recommended_action}

--- FULL CHAT TRANSCRIPT ---
${transcript}`,
        public: false
      },
      priority: summary.user_sentiment === 'Frustrated' ? 'high' : 'normal',
      requester: {
        name: user.name,
        email: user.email
      }
    }
  };

  return ticketData;
}

Setting public: false on the comment ensures this background information stays internal. The customer will not see this dump.

Step 3: Send the Ticket via Zendesk API

Send the formatted payload to Zendesk using basic authentication with your API token.

const axios = require('axios');

async function sendToZendesk(ticketPayload) {
  const zendeskDomain = process.env.ZENDESK_SUBDOMAIN;
  const authHeader = Buffer.from(
    `${process.env.ZENDESK_EMAIL}/token:${process.env.ZENDESK_API_TOKEN}`
  ).toString('base64');

  try {
    const response = await axios.post(
      `https://${zendeskDomain}.zendesk.com/api/v2/tickets.json`,
      ticketPayload,
      {
        headers: {
          'Authorization': `Basic ${authHeader}`,
          'Content-Type': 'application/json'
        }
      }
    );
    return response.data.ticket.id;
  } catch (error) {
    console.error('Failed to create Zendesk ticket:', error.response?.data || error.message);
    throw error;
  }
}

If you are setting up larger automated routing logic across platforms, exploring workflow automation techniques can help you link custom webhooks directly to ticket routing tables without hardcoding everything.

Step 4: Transfer Live Web Chat State

If you use Zendesk Messaging or Web Widget for real time chat rather than email tickets, update the metadata directly on the active web session before passing control.

Use the Zendesk Messaging JS API on your web frontend:

zE('messenger', 'set', 'conversationFields', [
  { id: '12345678', value: summary.customer_intent },
  { id: '87654321', value: summary.user_sentiment }
]);

zE('messenger', 'open');

This populates custom ticket fields on the agent sidebar instantly when the live workspace picks up the chat thread.

Expected Outcomes

Once this protocol is active, your support pipeline will behave predictably:

When an AI bot triggers a handoff, the backend executes the summary prompt in under two seconds.
A Zendesk ticket is created automatically, populated with structured metadata and private notes.
The human support agent receives the chat or ticket with full context already displayed on their sidebar.
Average handle time drops significantly, and customers never have to repeat their account numbers or problem details.

If you need extra engineering horsepower to build customized AI tools or integrations like this for your operations, Gaper can match you with vetted developers who specialize in AI pipelines and backend infrastructure.

How to Build a Hybrid Workflow Engine: Combining Finite State Machines with LLM Reasoning Nodes

MUHAMMAD MUSADIQ — Thu, 30 Jul 2026 04:17:00 +0000

Building AI applications purely on LLM chains usually ends in tears. One day, the model outputs crisp JSON, and the next, it returns a polite paragraph that breaks your downstream code. On the flip side, traditional state machines are bulletproof but rigid—they fail the moment a user submits ambiguous text or an unpredictable input.

I spent weeks debugging broken multi-prompt pipelines before realizing the solution was simple: we need to fuse deterministic state machines with LLM reasoning nodes. By using a durable workflow engine like Temporal to manage state transitions while letting LLMs make bounded decisions, you get reliable, deterministic AI systems that won't crash in production.

Here is how you can build a hybrid workflow engine from scratch.

A state machine needs explicit states, valid transitions, and strict rules. An LLM node takes raw context, evaluates it, and returns a single decision from a pre-defined set of valid actions.

In a hybrid setup, the state machine handles orchestration, state persistence, retries, and execution limits. The LLM acts purely as a routing function inside a single state. The LLM never updates global state directly. It simply suggests the next transition based on structured data.

If you are scaling enterprise systems, setting up robust workflow automation early prevents massive tech debt later.

+-----------------------------------------------------------+
|                    Temporal Workflow                      |
|                                                           |
|  [State: DRAFT] ---> [Activity: LLM Evaluation Node]      |
|                                |                          |
|                                v                          |
|                      Structured Decision                  |
|                        /              \                   |
|                       v                v                  |
|          [State: REQUIRES_HUMAN]   [State: APPROVED]      |
+-----------------------------------------------------------+

Step 2: Set Up Temporal and Dependencies

We will use Temporal for durable execution and Python for our workflow logic. First, spin up a local Temporal server using the Temporal CLI or Docker.

Install the required Python packages:

pip install temporalio openai pydantic

Make sure your Temporal server is running locally on port 7233.

Step 3: Define Rigid State Models with Pydantic

To enforce deterministic AI behavior, the LLM must return strict schema outputs. Never let the LLM output freeform text to control workflow state.

Create a file named schemas.py:

from enum import Enum
from pydantic import BaseModel, Field

class WorkflowState(str, Enum):
    DRAFT = "DRAFT"
    NEEDS_REVISION = "NEEDS_REVISION"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"

class LLMDecision(str, Enum):
    APPROVE = "APPROVE"
    REQUEST_CHANGES = "REQUEST_CHANGES"
    REJECT = "REJECT"

class EvaluationResult(BaseModel):
    decision: LLMDecision
    reasoning: str = Field(description="Brief reason for the decision")
    confidence_score: float = Field(description="Confidence between 0.0 and 1.0")

Step 4: Build the LLM Reasoning Activity

Temporal activities handle side effects like API calls or database operations. If an LLM call fails or hits a rate limit, Temporal automatically handles retries without restarting the whole workflow.

When teams start building complex AI agents, keeping side effects inside isolated activities is the single best decision they can make.

Create activities.py:

import json
import os
from openai import OpenAI
from temporalio import activity
from schemas import EvaluationResult, LLMDecision

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@activity.defn
async def evaluate_document_activity(document_text: str) -> EvaluationResult:
    prompt = f"""
    Analyze the following text for compliance and quality.
    Select APPROVE if it meets standard criteria.
    Select REQUEST_CHANGES if minor edits are needed.
    Select REJECT if it violates basic guidelines.

    Document Text:
    {document_text}
    """

    response = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a deterministic logic node in a workflow engine."},
            {"role": "user", "content": prompt}
        ],
        response_format=EvaluationResult
    )

    return response.choices[0].message.parsed

Step 5: Implement the State Machine Workflow

Now write the workflow that processes state transitions based on the output of the LLM node. Temporal guarantees that workflow state persists across crashes, server restarts, or network failures.

Create workflows.py:

from datetime import timedelta
from temporalio import workflow
from schemas import WorkflowState, LLMDecision, EvaluationResult

with workflow.unsafe.imports_passed_through():
    from activities import evaluate_document_activity

@workflow.defn
class DocumentApprovalWorkflow:
    def __init__(self):
        self.current_state = WorkflowState.DRAFT
        self.history = []

    @workflow.run
    async def run(self, document_text: str) -> dict:
        self.history.append(f"Started in state: {self.current_state}")

        # Execute LLM Reasoning Node
        result: EvaluationResult = await workflow.execute_activity(
            evaluate_document_activity,
            document_text,
            start_to_close_timeout=timedelta(seconds=30)
        )

        # Enforce State Machine Transitions based on LLM output
        if result.decision == LLMDecision.APPROVE and result.confidence_score > 0.8:
            self.current_state = WorkflowState.APPROVED
        elif result.decision == LLMDecision.REQUEST_CHANGES:
            self.current_state = WorkflowState.NEEDS_REVISION
        else:
            # Low confidence or explicit reject routes to REJECTED
            self.current_state = WorkflowState.REJECTED

        self.history.append(f"Transitioned to state: {self.current_state} | Reason: {result.reasoning}")

        return {
            "final_state": self.current_state,
            "confidence": result.confidence_score,
            "history": self.history
        }

Step 6: Create the Worker and Trigger the Workflow

To run the workflow, set up a worker process to listen on a Temporal task queue.

Create run_worker.py:

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from activities import evaluate_document_activity
from workflows import DocumentApprovalWorkflow

async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="hybrid-workflow-queue",
        workflows=[DocumentApprovalWorkflow],
        activities=[evaluate_document_activity],
    )
    print("Worker started. Listening for tasks...")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

In a separate terminal, trigger the execution using execute_workflow.py:

import asyncio
from temporalio.client import Client
from workflows import DocumentApprovalWorkflow

async def main():
    client = await Client.connect("localhost:7233")

    sample_doc = "The policy document outlines terms for user data retention. All data is deleted after 30 days."

    result = await client.execute_workflow(
        DocumentApprovalWorkflow.run,
        sample_doc,
        id="doc-approval-001",
        task_queue="hybrid-workflow-queue",
    )

    print("Workflow Execution Result:")
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

Run the worker terminal first, then run the execution script. You will see Temporal step through the workflow states deterministically, using the LLM output strictly as a routing signal.

Moving Beyond Simple Logic

Building hybrid engines keeps your production code sane. State machines provide observability, deterministic retries, and guardrails. LLMs supply flexible evaluation and contextual understanding. Mixing them prevents unpredictable loop failures and runaway API bills.

If you are expanding your infrastructure and need expert developers to scale these systems, Gaper connects companies with vetted software engineers who specialize in production ready workflow orchestration and AI implementations.

How to Implement Human-in-the-Loop Approval Gates for Risky AI Agent Tool Calls

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 05:03:10 +0000

Autonomous AI agents are great until they drop a database table or send an unintended email to thousands of clients. Letting an LLM execute arbitrary actions without human validation is a fast track to production incidents. To keep your systems safe, you need an approval gate that pauses high-risk tool calls and waits for explicit human confirmation before executing sensitive operations.

Python 3.9 or higher installed on your machine.
Basic familiarity with Python functions and dictionaries.
No external libraries required, standard Python libraries like functools and typing work fine.

Step 1: Flag Risky Tools with Metadata

Start by defining custom tool decorators or metadata to tag functions as high risk. You do not need complex frameworks to do this. A simple decorator that attaches a requires_approval attribute to sensitive functions keeps your code clean.

import functools

def ai_tool(requires_approval: bool = False):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        wrapper.requires_approval = requires_approval
        wrapper.is_ai_tool = True
        return wrapper
    return decorator

# Example tools
@ai_tool(requires_approval=False)
def search_knowledge_base(query: str) -> str:
    return f"Results for: {query}"

@ai_tool(requires_approval=True)
def delete_user_account(user_id: str) -> str:
    return f"User {user_id} successfully deleted."

Step 2: Build the Interceptor Routine

Next, write an execution layer that inspects the target tool before running it. If the function has requires_approval=True, the wrapper suspends execution and requests user consent via console input, API callback, or Slack notification.

Here is a lightweight CLI implementation:

def execute_tool(tool_func, *args, **kwargs):
    tool_name = tool_func.__name__
    needs_review = getattr(tool_func, "requires_approval", False)

    if needs_review:
        print(f"\n[APPROVAL REQUIRED] Agent requested to call: {tool_name}")
        print(f"Arguments: args={args}, kwargs={kwargs}")

        user_input = input("Approve execution? (y/N): ").strip().lower()
        if user_input != 'y':
            print(f"[BLOCKED] Execution of {tool_name} was aborted by user.")
            return {"status": "rejected", "reason": "User denied approval"}

    print(f"[EXECUTING] Running {tool_name}...")
    result = tool_func(*args, **kwargs)
    return {"status": "success", "result": result}

This pattern works well whether you are building custom AI agent development pipelines or setting up complex enterprise rules. For instance, in sensitive fields like AI for Accounting & Finance, automated fund transfers or invoice modifications should always hit an explicit guardrail before moving real money.

Step 3: Handle Agent Responses and Execution

Now wire the tool executor directly into your agent call loop. When the model selects a tool, route the call through your execute_tool router rather than calling the tool directly.

# Simulated model tool calls coming from an agent framework
tool_calls_queue = [
    {"func": search_knowledge_base, "args": ("password reset steps",)},
    {"func": delete_user_account, "kwargs": {"user_id": "usr_9981"}}
]

for call in tool_calls_queue:
    func = call["func"]
    args = call.get("args", ())
    kwargs = call.get("kwargs", {})

    response = execute_tool(func, *args, **kwargs)
    print("Execution output:", response)

Integrating human approval gates like this forms the backbone of reliable workflow automation, giving teams confidence to deploy autonomous systems without giving up control over dangerous actions.

Expected Results

When running the full script, standard read operations execute immediately, while high-risk functions prompt for explicit approval:

[EXECUTING] Running search_knowledge_base...
Execution output: {'status': 'success', 'result': 'Results for: password reset steps'}

[APPROVAL REQUIRED] Agent requested to call: delete_user_account
Arguments: args=(), kwargs={'user_id': 'usr_9981'}
Approve execution? (y/N): n
[BLOCKED] Execution of delete_user_account was aborted by user.
Execution output: {'status': 'rejected', 'reason': 'User denied approval'}

If you answer y, the function executes and returns the success output. If you enter n or anything else, the process halts safely and returns a clear refusal message to the agent so it can handle the failure gracefully.

If you need help scaling agent architectures or embedding custom governance controls into your production stack, the team at Gaper can connect you with experienced engineers who build secure AI software.

How to Build a Multi-Tenant Data Isolation Layer for Enterprise AI Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 05:00:48 +0000

Deploying enterprise AI agents across multiple clients quickly becomes dangerous if your retrieval pipeline lacks strict tenant boundaries. Standard vector searches only evaluate semantic similarity, so without hard filtering, your agent can easily pull data from Tenant A to answer a prompt from Tenant B. I learned this the hard way when a test agent leaked a competitor's internal notes into a chat response during an early staging build. To fix this, you need a dedicated isolation layer that intercepts every agent tool call, injects validated tenant IDs, and enforces strict data boundaries at the database level.

Before writing code, make sure you have the following ready:

Python 3.10 or higher installed
A vector database supporting payload metadata filtering, such as Qdrant or PostgreSQL with pgvector
FastAPI or Flask for your web backend
An identity provider issuing JWTs that contain tenant metadata

Step 1: Create a Thread-Safe Tenant Context

Your system needs a reliable way to track who is making the request across asynchronous function calls. Do not rely on the LLM to pass tenant IDs in tool parameters. The model can hallucinate parameters or get tricked via prompt injection.

Use Python's contextvars module to store the active tenant ID securely in memory for each request context.

import contextvars

# Context variable scoped to the current execution thread or async task
_tenant_context = contextvars.ContextVar("tenant_id", default=None)

def set_tenant(tenant_id: str):
    return _tenant_context.set(tenant_id)

def get_tenant() -> str:
    tenant_id = _tenant_context.get()
    if not tenant_id:
        raise PermissionError("Access denied. No active tenant context found.")
    return tenant_id

Step 2: Build Middleware to Extract and Set Context

Intercept incoming API calls before they ever reach your agent runtime. Read the user's JWT, extract the verified tenant identifier, and store it in your context variable.

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

def verify_and_parse_jwt(token: str) -> dict:
    # Decode and validate your JWT here using your auth provider's public key
    # Returning mock data for demonstration
    return {"tenant_id": "tenant_company_abc"}

@app.middleware("http")
async def tenant_isolation_middleware(request: Request, call_next):
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing authorization header.")

    raw_token = auth_header.split(" ")[1]
    claims = verify_and_parse_jwt(raw_token)

    # Store tenant context for the duration of this request lifecycle
    token_ref = set_tenant(claims["tenant_id"])
    try:
        response = await call_next(request)
        return response
    finally:
        # Clear context after request finishes
        _tenant_context.reset(token_ref)

Step 3: Inject System Filters into Vector Store Queries

When your agent triggers a retrieval tool, your vector database wrapper must automatically attach metadata filters. If you are building custom AI workflows, teams often use custom tooling layers. You can explore specialized patterns at AI Agent Development to see how agents scale in production.

Here is how you inject the tenant filter programmatically:

def search_knowledge_base(query_vector: list, top_k: int = 5):
    # Fetch tenant ID directly from global request context
    active_tenant = get_tenant()

    # Enforce metadata filter regardless of what the LLM requested
    tenant_filter = {
        "must": [
            {"key": "tenant_id", "match": {"value": active_tenant}}
        ]
    }

    # Example Qdrant client query
    results = vector_client.search(
        collection_name="enterprise_knowledge",
        query_vector=query_vector,
        query_filter=tenant_filter,
        limit=top_k
    )
    return results

Step 4: Wrap Custom Agent Tools with Security Decorators

Agents usually do more than vector lookups. They execute SQL queries, send webhooks, or interact with external services. Create a decorator that wraps all agent tools and forces tenant check verification.

If your setup spans multiple legacy systems, reviewing AI implementation and consulting strategies can help map out cross-system data boundaries.

from functools import wraps

def enforce_tenant_boundary(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        current_tenant = get_tenant()
        print(f"[AUDIT LOG] Executing {func.__name__} for Tenant: {current_tenant}")

        # Inject verified tenant ID into tool parameters explicitly
        kwargs["tenant_id"] = current_tenant
        return func(*args, **kwargs)
    return wrapper

@enforce_tenant_boundary
def fetch_customer_invoices(customer_id: str, tenant_id: str = None):
    # The database query MUST use the injected tenant_id parameter
    query = "SELECT * FROM invoices WHERE customer_id = %s AND tenant_id = %s"
    return db.execute(query, (customer_id, tenant_id))

Expected Outcomes

After deploying this isolation layer, your system achieves several clear security guarantees:

Data leakage between tenants becomes impossible at the tool layer because filters are appended outside the LLM's control.
Prompt injection attacks trying to override tenant settings will fail completely.
Audit logs automatically capture every tool execution alongside a verified tenant identifier.

If you need extra engineering horsepower to design or implement secure AI architectures, Gaper connects companies with vetted developers ready to build enterprise software.

Building Model-Agnostic AI Agents: How to Implement Runtime Fallbacks Between OpenAI, Claude, and Gemini

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:55:11 +0000

Relying on a single LLM provider for a production application is risky. API rate limits, temporary outages, and unexpected latency spikes can bring down your application without warning. When your primary provider goes offline, your users shouldn't be stuck staring at a loading spinner. You need a system that automatically routes requests to backup providers like Claude or Gemini, without breaking your agent's state or changing your output format.

Here is how to build a lightweight, model-agnostic fallback handler in Python to keep your agents running smoothly.

Before starting, make sure you have:

Python 3.10 or higher installed.
API keys for OpenAI, Anthropic, and Google Gemini.
The provider SDKs installed (openai, anthropic, google-genai).

If you are building complex systems, setting up robust AI agent development patterns early saves you dozens of hours in maintenance down the road.

Step 1: Normalize Prompts and Outputs

Different providers expect different payload formats. OpenAI uses a messages list with role and content. Anthropic uses a similar system but handles system prompts as a top-level parameter. Gemini uses its own content structure.

Create a single standard format for your application.

from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class ChatMessage:
    role: str # "system", "user", or "assistant"
    content: str

@dataclass
class LLMResponse:
    content: str
    provider: str
    model: str

Step 2: Build Generic Provider Wrappers

Wrap each client in a standard interface. Each function should accept your ChatMessage list and return your LLMResponse.

import os
from openai import OpenAI
from anthropic import Anthropic
import google.genai as genai

class ProviderAdapter:
    def __init__(self):
        self.openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.anthropic_client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))

    def call_openai(self, messages: List[ChatMessage], model="gpt-4o") -> LLMResponse:
        formatted = [{"role": m.role, "content": m.content} for m in messages]
        res = self.openai_client.chat.completions.create(model=model, messages=formatted)
        return LLMResponse(
            content=res.choices[0].message.content,
            provider="openai",
            model=model
        )

    def call_claude(self, messages: List[ChatMessage], model="claude-3-5-sonnet-20241022") -> LLMResponse:
        system_prompt = ""
        user_messages = []

        for m in messages:
            if m.role == "system":
                system_prompt += m.content + "\n"
            else:
                user_messages.append({"role": m.role, "content": m.content})

        res = self.anthropic_client.messages.create(
            model=model,
            max_tokens=1024,
            system=system_prompt.strip(),
            messages=user_messages
        )
        return LLMResponse(
            content=res.content[0].text,
            provider="anthropic",
            model=model
        )

    def call_gemini(self, messages: List[ChatMessage], model="gemini-2.5-flash") -> LLMResponse:
        # Flatten system and user messages into a simple text prompt for basic compatibility
        prompt_text = "\n".join([f"{m.role}: {m.content}" for m in messages])
        res = self.gemini_client.models.generate_content(
            model=model,
            contents=prompt_text
        )
        return LLMResponse(
            content=res.text,
            provider="gemini",
            model=model
        )

Step 3: Write the Fallback Router Loop

Now build the fallback router. It loops through your preferred order of models. If one fails due to an API error, timeout, or rate limit, it catches the exception, logs a warning, and moves to the next model immediately.

import logging

logging.basicConfig(level=logging.INFO)

class ResilientAgent:
    def __init__(self):
        self.adapter = ProviderAdapter()
        # Define fallback hierarchy: (provider_name, model_name)
        self.fallback_chain = [
            ("openai", "gpt-4o"),
            ("anthropic", "claude-3-5-sonnet-20241022"),
            ("gemini", "gemini-2.5-flash")
        ]

    def execute(self, messages: List[ChatMessage]) -> LLMResponse:
        for provider, model in self.fallback_chain:
            try:
                logging.info(f"Attempting execution with {provider} ({model})...")

                if provider == "openai":
                    return self.adapter.call_openai(messages, model=model)
                elif provider == "anthropic":
                    return self.adapter.call_claude(messages, model=model)
                elif provider == "gemini":
                    return self.adapter.call_gemini(messages, model=model)

            except Exception as e:
                logging.warning(f"Provider {provider} failed with error: {e}. Trying fallback...")

        raise RuntimeError("All LLM providers failed. Check API status and keys.")

This pattern is useful when building mission critical system integration and workflow automation, where uptime is mandatory for daily business operations.

Step 4: Test Your Fallback Flow

You can run a quick test script to make sure the fallback triggers when the primary provider encounters an issue.

if __name__ == "__main__":
    agent = ResilientAgent()

    conversation = [
        ChatMessage(role="system", content="You are a helpful assistant."),
        ChatMessage(role="user", content="Explain runtime fallbacks in one sentence.")
    ]

    # Simulating a failure by supplying a bogus key for OpenAI in env variables
    # will force the agent to skip OpenAI and hit Claude or Gemini automatically.

    try:
        response = agent.execute(conversation)
        print("\n--- Success ---")
        print(f"Provider Used: {response.provider}")
        print(f"Model Used: {response.model}")
        print(f"Response: {response.content}")
    except Exception as err:
        print(f"Execution failed completely: {err}")

Expected Outcomes

When you run this architecture, your logs will show the system attempting your primary provider first. If OpenAI hits a 500 server error or a 429 rate limit, the loop catches the exception in milliseconds and switches to Claude or Gemini.

The calling application receives a standard LLMResponse regardless of which vendor actually processed the request. Your downstream data parsing remains identical, keeping your application stable even during major cloud vendor outages.

If you need help scaling this pattern across multi-agent systems or enterprise infrastructure, the team at Gaper provides senior engineers who specialize in production ready AI systems.

How to Manage Long-Running State and Asynchronous Callbacks in Autonomous Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:51:10 +0000

Building autonomous agents that handle simple, single-turn prompts is easy. The real headache starts when your agent has to wait three days for a human manager to approve an action, or ten minutes for an external API to process a batch of files. If your agent holds its context in memory while waiting, a single server restart wipes out everything. I learned this the hard way when a deployment killed dozens of active workflows mid-execution, leaving users stranded with half-finished tasks.

To build reliable long running agents, you need a system that decouples execution from memory. Your agent must be able to pause, write its full context to storage, yield control, and wake up only when a trigger or callback arrives.

Here is how you can implement durable agent state management and asynchronous callbacks step by step.

Traditional agent scripts run in a continuous loop: think, act, observe, repeat. This works fine for quick scripts, but fails for long-running processes.

[ Traditional Agent Loop ]
Agent Start -> Prompt LLM -> Call Tool -> Wait for API (Blocking) -> Process -> Done
(If process crashes during Wait, state is lost forever)

[ Event-Driven Durable Agent ]
Agent Start -> Prompt LLM -> Save State to DB -> Suspend Process
                                                        |
   Webhook / Human Input arrives later -----------------+
                                                        v
                               Resume Process -> Load State -> Complete Task

To handle asynchronous callbacks, we break this single process into three main parts:

State Persistence: Saving the memory, scratchpad, and conversation history to a database.
Execution Suspension: Halting execution cleanly without holding open HTTP connections or thread locks.
Event Rehydration: Rebuilding the agent's full state when a webhook or user input triggers a wake-up call.

If you are scaling complex AI agent development, getting this foundation right early saves hundreds of hours of debugging downstream.

Step 1: Define a Serializable Agent State Schema

You cannot save raw LLM instances or active client connections to a database. You must serialize the agent state into a structured schema containing the prompt history, tool call outputs, pending callbacks, and current status.

Here is a Pydantic schema using Python that captures what a durable execution setup needs:

from enum import Enum
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
import datetime

class AgentStatus(str, Enum):
    RUNNING = "running"
    SUSPENDED = "suspended"
    COMPLETED = "completed"
    FAILED = "failed"

class AgentState(BaseModel):
    thread_id: str
    status: AgentStatus = AgentStatus.RUNNING
    system_prompt: str
    messages: List[Dict[str, Any]] = Field(default_factory=list)
    pending_callback_id: Optional[str] = None
    context_data: Dict[str, Any] = Field(default_factory=dict)
    created_at: str = Field(default_factory=lambda: datetime.datetime.utcnow().isoformat())
    updated_at: str = Field(default_factory=lambda: datetime.datetime.utcnow().isoformat())

This model stores everything necessary to rebuild the agent environment. The pending_callback_id field tracks external async actions like email responses, third-party webhook processing, or human approvals.

Step 2: Implement Persistent Storage for Agent State Management

Next, build a persistence repository. SQLite works great for local testing, while PostgreSQL or Redis fits production workloads.

This simple repository pattern manages durable state transitions:

import sqlite3
import json

class StateRepository:
    def __init__(self, db_path: str = "agent_state.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS states (
                    thread_id TEXT PRIMARY KEY,
                    status TEXT,
                    pending_callback_id TEXT,
                    payload TEXT
                )
            """)
            conn.commit()

    def save(self, state: AgentState):
        state.updated_at = datetime.datetime.utcnow().isoformat()
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                """
                INSERT INTO states (thread_id, status, pending_callback_id, payload)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(thread_id) DO UPDATE SET
                    status=excluded.status,
                    pending_callback_id=excluded.pending_callback_id,
                    payload=excluded.payload
                """,
                (
                    state.thread_id,
                    state.status.value,
                    state.pending_callback_id,
                    json.dumps(state.dict())
                )
            )
            conn.commit()

    def load(self, thread_id: str) -> Optional[AgentState]:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT payload FROM states WHERE thread_id = ?", (thread_id,))
            row = cursor.fetchone()
            if row:
                return AgentState(**json.loads(row[0]))
            return None

With persistent storage ready, the agent can stop execution entirely while waiting for asynchronous events.

Step 3: Handle Asynchronous Callbacks and Suspensions

When your agent hits an async barrier, such as waiting for a human review or calling a long-running batch extraction API, it should emit an external request, register a unique callback key, save state, and exit.

Here is an example showing how an agent suspends its flow:

import uuid

class AgentEngine:
    def __init__(self, repo: StateRepository):
        self.repo = repo

    def execute_human_approval_step(self, thread_id: str, action_details: str):
        state = self.repo.load(thread_id)
        if not state:
            raise ValueError("Thread not found")

        # Create a unique callback listener token
        callback_id = f"cb_{uuid.uuid4().hex[:8]}"

        # Record action intent into conversation history
        state.messages.append({
            "role": "assistant",
            "content": f"Requesting human approval for action: {action_details}"
        })

        # Transition state to suspended
        state.status = AgentStatus.SUSPENDED
        state.pending_callback_id = callback_id

        # Save state to storage
        self.repo.save(state)

        print(f"[Agent Suspended] Task waiting for callback token: {callback_id}")
        return callback_id

Instead of blocking thread execution, this design frees memory and CPU resources. It is especially useful when integrating complex backend tasks with workflow automation pipelines that spans across external services.

Step 4: Resume Execution on Callback Arrival

Now, set up a handler to accept incoming callbacks, find the matching agent execution state, inject the incoming payload, and resume processing.

    def handle_callback(self, callback_id: str, payload: Dict[str, Any]):
        # Search state database for matching callback token
        with sqlite3.connect(self.repo.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT thread_id FROM states WHERE pending_callback_id = ?", (callback_id,))
            row = cursor.fetchone()

        if not row:
            raise ValueError(f"No agent suspended for callback ID: {callback_id}")

        thread_id = row[0]
        state = self.repo.load(thread_id)

        # Update thread state with incoming payload
        state.messages.append({
            "role": "user",
            "content": f"Callback received with data: {json.dumps(payload)}"
        })

        # Clear callback lock and set back to running
        state.pending_callback_id = None
        state.status = AgentStatus.RUNNING
        self.repo.save(state)

        print(f"[Agent Resumed] Processing thread: {thread_id}")
        return self.continue_agent_loop(state)

    def continue_agent_loop(self, state: AgentState):
        # Continue reasoning loop with updated state context
        state.messages.append({
            "role": "assistant",
            "content": "Received callback approval. Proceeding with execution."
        })
        state.status = AgentStatus.COMPLETED
        self.repo.save(state)
        return state

Step 5: Put It Together and Run an Simulation

Here is a full demonstration of the cycle: starting an agent task, pausing for external input, and resuming execution.

if __name__ == "__main__":
    repo = StateRepository()
    engine = AgentEngine(repo)

    # Initialize a new agent run
    thread_id = f"session_{uuid.uuid4().hex[:6]}"
    initial_state = AgentState(
        thread_id=thread_id,
        system_prompt="You are an autonomous assistant handling invoice operations.",
        messages=[{"role": "user", "content": "Process refund for Order #9482"}]
    )
    repo.save(initial_state)

    # Agent runs until it needs external authorization
    callback_token = engine.execute_human_approval_step(
        thread_id=thread_id, 
        action_details="Refund $450 to customer"
    )

    # At this point, the application process can close, restart, or crash.
    # The agent state remains safe inside SQLite.

    # Later, an incoming webhook or API call fires this endpoint handler:
    print("\n... Waiting for external webhook callback ...\n")

    incoming_webhook_payload = {
        "approved": True,
        "approver": "manager@company.com"
    }

    final_state = engine.handle_callback(callback_token, incoming_webhook_payload)
    print(f"Final Agent Status: {final_state.status.value}")

Running this code prints:

[Agent Suspended] Task waiting for callback token: cb_a1b2c3d4

... Waiting for external webhook callback ...

[Agent Resumed] Processing thread: session_e5f6g7
Final Agent Status: completed

Best Practices for Production Durable Execution

When building production grade systems for async agent workflows, keep these tips in mind:

Use Dedicated Task Orchestrators: While SQLite works for basic usage, engines like Temporal, Prefect, or AWS Step Functions handle re-runs, retries, and time-outs much better.
Store Minimal Payload Context: Do not save large binary outputs directly inside the agent execution payload. Store them in S3 or object storage, then pass references into the agent state object.
Handle Timeouts: Always append expiration timers to suspended states. If a callback never returns, a fallback background job should wake the agent up to handle the error condition gracefully.

Building long running agents requires shifting from simple synchronous loops to robust, stateful architectures. If your team needs hand building reliable agent infrastructures, our engineers at Gaper can help design state management patterns suited for enterprise scale.

Why Fine-Tuning Isn't Enough: Combining RAG, Tools, and Runtime Guardrails for Production Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:48:13 +0000

Last year, I watched an engineering team spend six months and tens of thousands of dollars fine-tuning an open-source model on their company's internal documentation. They were convinced it would replace their front-line customer support stack. On launch day, the bot gave polite, beautifully formatted answers that were completely, spectacularly wrong. It hallucinated pricing plans that never existed and cited policy documents from three years ago.

That project was scrapped two weeks later, but the lesson stuck with me. We are watching hundreds of teams make the exact same mistake right now. They treat weight training like a magic database.

Fine tuning is great for style, tone, and syntax. If you want a model to sound like a medical professional, adhere to strict JSON formats, or master a rare programming syntax, fine tuning works well.

Fine tuning is not memory. It is a static snapshot.

The moment you finish training a model checkpoint, your data starts decaying. When your product team updates pricing tomorrow, your shiny fine-tuned model becomes an operational hazard. Re-training every time your business logic changes is slow, expensive, and fragile. You end up fighting catastrophic forgetting, where the model forgets general reasoning skills just to memorize a few API endpoints.

The Real Architecture of Production AI Agents

Building reliable production AI agents requires a modular software architecture. You cannot rely on a single neural network to act as database, logic engine, API controller, and security guard all at once. Basic software engineering rules still apply here. We need clear separation of concerns.

A production agent system rests on three distinct pillars that sit outside the base model weights: Retrieval-Augmented Generation (RAG), external tool integration, and runtime guardrails.

RAG and Tools: Giving the Model Sight and Motion

Think of a large language model as an intelligent consultant with temporary amnesia. RAG is the folder of fresh background context you place on their desk before asking a question. By decoupling knowledge from model weights, you can update your vector stores and knowledge graphs instantly. When a policy changes, you update your database, not your training scripts.

Tools give the agent hands. An LLM should never calculate loan amortization or query a database through raw token prediction. It will fail. Instead, the model should decide which tool to call, pass structured parameters, and let a deterministic script do the real work.

When you need complex business logic or enterprise workflow automation, tools let the model orchestrate existing, battle-tested software rather than reinventing execution logic on the fly.

Runtime Guardrails Are Not Optional

Even with pristine context and reliable tools, models drift. They get confused by long prompt histories. They attempt invalid function calls. They break character when a user tries a basic prompt injection.

This is where runtime guardrails come in. You need deterministic software layers wrapping your model calls.

Input guardrails inspect incoming user prompts for malicious intent, topical relevance, and Pydantic schema alignment before the model ever sees them. Output guardrails parse the raw text response, check for policy violations, validate output structures, and verify facts against the original retrieved context. If the model strays, the system catches it, logs the anomaly, and triggers a retry loop or a safe fallback mechanism.

Fine tuning might take your output format quality from 70% to 90%. Runtime guardrails and schema validation are what take your system reliability from 90% to 99.9%.

Where Fine Tuning Actually Belongs

Fine tuning isn't dead, but its role in the stack has shifted.

Instead of using fine tuning to teach your model facts, use it to lower latency and cut token overhead. If your system prompt requires 2,000 tokens of few-shot examples just to force a specific output schema, that is a great candidate for fine tuning. Train a smaller, faster model on that specific format, then put that model inside a runtime loop powered by dynamic RAG, structured tool calling, and guardrails.

Stop expecting a single weight file to solve architectural problems. Build robust systems around your models instead.

If you want to build durable systems without spending months down the wrong architectural path, take a look at the team at Gaper. Their engineers specialize in building production-grade agent frameworks and providing practical AI consulting to help companies deploy models that actually work in the real world.

Designing Scoped API Authorization and RBAC for Autonomous Support Agents

MUHAMMAD MUSADIQ — Wed, 29 Jul 2026 04:44:34 +0000

Autonomous support agents need access to internal APIs to issue refunds, update accounts, and retrieve user logs. However, giving an AI model direct access to your database or an admin API key poses a massive risk. If a user manipulates the agent through prompt injection, the agent could wipe a database or issue unauthorized payments. To prevent this, you need a security framework that restricts what an agent can do based on the authenticated user, the agent's assigned role, and strict API scopes.

Before building your authorization layer, make sure you have:

An API gateway or backend service running Node.js, Python, or Go.
An OAuth 2.0 Identity Provider supporting RFC 8693 token exchange.
An LLM framework set up for tool use or function calling.
Basic understanding of RBAC and JSON Web Tokens (JWTs).

Step 1: Define Scopes for Every Agent Tool

Never expose generic API endpoints to your AI tools. Instead, break your backend capabilities into explicit, low-privilege scopes.

Instead of giving your agent access to a generic POST /api/v1/payments route, create dedicated tool abstractions with specific scopes:

read:orders: View order history.
read:user_profile: View basic customer info.
write:refund_limited: Issue refunds up to a specific threshold, like $50.

Map these scopes directly to the tools in your agent's system prompt or function definitions. If the agent does not need to edit a user profile, its token context should not even possess the scope to do so.

Step 2: Implement OAuth Delegation for Agent Contexts

An agent should never act with full system admin rights. Use token delegation so the agent inherits the permissions of the user requesting help, constrained by the agent's own role.

When a customer starts a chat session:

Authenticate the customer and issue a short lived user token.
Request a delegated token for the support agent service using token exchange (RFC 8693).
The returned token contains the intersection of user permissions and agent permissions.

If a non-admin user asks the agent to delete an account, the delegated token won't have the delete:account scope. Even if prompt injection succeeds, the API gateway will block the tool call with a 403 Forbidden. If you are building complex integration pipelines, setting up clean workflow automation patterns helps keep token exchanges reliable across services.

Step 3: Add API Gateway Tool Authorization Middleware

Validate incoming tool execution requests at your API gateway or middleware, not inside the LLM logic. Treat the agent as an untrusted client.

Here is a simple Python example using FastAPI to enforce scope checks on a tool execution endpoint:

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

app = FastAPI()
security = HTTPBearer()

def verify_agent_scope(required_scope: str):
    def dependency(credentials: HTTPAuthorizationCredentials = Security(security)):
        token = credentials.credentials
        try:
            payload = jwt.decode(token, "YOUR_SECRET_KEY", algorithms=["HS256"])
            scopes = payload.get("scopes", []).split(" ")
            if required_scope not in scopes:
                raise HTTPException(status_code=403, detail="Insufficient tool permissions")
            return payload
        except jwt.PyJWTError:
            raise HTTPException(status_code=401, detail="Invalid token")
    return dependency

@app.post("/tools/issue-refund")
def issue_refund(amount: float, payload: dict = Depends(verify_agent_scope("write:refund_limited"))):
    if amount > 50.0:
        raise HTTPException(status_code=400, detail="Amount exceeds automated limit")
    return {"status": "success", "refunded": amount}

This ensures that even if the agent hallucinates a function call with invalid parameters, your backend enforces hard security limits.

Step 4: Require Step-Up Authorization for High-Risk Operations

For operations that exceed normal bounds, require step-up authorization.

If a customer asks for a $200 refund, the agent tool should return an approval_required status message instead of executing the API call directly. The backend generates an approval task sent to a human support queue. Once the human approves, a temporary, single-use token is generated to complete the transaction.

If you want to skip building these RBAC architectures from scratch, custom AI agent development services can help structure secure tool infrastructure right out of the box.

Expected Outcomes

Once you implement this RBAC design, you can expect:

Strong defense against prompt injection attacks that attempt unauthorized tool execution.
Granular audit logs showing which user session requested an action and which agent context executed it.
Reduced operational risk by enforcing strict value ceilings on autonomous tasks.

If you are looking for guidance on implementing secure agent infrastructure or scaling custom models, check out Gaper for technical resources and engineering support.

How to Build an Immutable Audit Logging Pipeline for AI Agent Actions

MUHAMMAD MUSADIQ — Tue, 28 Jul 2026 05:27:58 +0000

title: How to Build an Immutable Audit Logging Pipeline for AI Agent Actions
published: false

description: Standard logs fall short when tracking autonomous AI agents. Learn how to build a tamper-proof, cryptographically sealed audit logging pipeline to guarantee compliance and securely record every agent action.

When an autonomous AI agent executes a bad database query or calls an external API with unexpected parameters, figuring out what happened is a nightmare. Standard application logs quickly fall short. They can be edited, rotated, or deleted, which breaks trust and makes AI compliance nearly impossible. If your agents handle real-world operations, you need tamper-proof audit logs that guarantee every step of an agent's execution is recorded, linked, and cryptographically sealed.

I learned this the hard way when an experimental customer support agent started firing off refund requests. Without a clean event history, I spent hours untangling state changes. In this guide, we will build a production-grade, immutable audit logging pipeline using Python, cryptographic hash chains, and append-only storage concepts.

An immutable audit trail needs three main features:

Structured payloads capturing inputs, outputs, tool calls, and state metadata.
Cryptographic hashing where each log entry contains the hash of the preceding entry, forming a tamper-evident chain.
Append-only storage that prevents modification or deletion of past records.

Here is how data flows through our observability pipeline:

[ Agent Action / Tool Call ] 
          │
          ▼
[ Log Interceptor / Context ]
          │
          ▼
[ Hash Chaining Engine ] (SHA-256 link to previous entry)
          │
          ▼
[ Append-Only Storage ] (PostgreSQL / S3 Object Lock)

Let's write the code step by step.

Step 1: Define the Log Entry and Cryptographic Engine

First, create a structured log entry schema. We will use standard SHA-256 hashes to link log entries together. If anyone alters an old log entry, the entire hash chain breaks, immediately signaling tampering.

Create a file named audit_chain.py:

import hashlib
import json
from datetime import datetime, timezone
from typing import Dict, Any, Optional

class AuditEntry:
    def __init__(self, agent_id: str, action: str, payload: Dict[str, Any], previous_hash: str = "0"):
        self.timestamp = datetime.now(timezone.utc).isoformat()
        self.agent_id = agent_id
        self.action = action
        self.payload = payload
        self.previous_hash = previous_hash
        self.hash = self.calculate_hash()

    def calculate_hash(self) -> str:
        # Create a deterministic string representation of the entry data
        data = {
            "timestamp": self.timestamp,
            "agent_id": self.agent_id,
            "action": self.action,
            "payload": self.payload,
            "previous_hash": self.previous_hash
        }
        serialized = json.dumps(data, sort_keys=True)
        return hashlib.sha256(serialized.encode('utf-8')).hexdigest()

    def to_dict(self) -> Dict[str, Any]:
        return {
            "timestamp": self.timestamp,
            "agent_id": self.agent_id,
            "action": self.action,
            "payload": self.payload,
            "previous_hash": self.previous_hash,
            "hash": self.hash
        }

This structure ensures that every entry binds itself to the previous state.

Step 2: Intercept Agent Execution

To record agent tracing data without cluttering your core agent logic, wrap tool calls and LLM invocations using Python decorators or middleware.

If you are building custom agents for enterprise workflows like those supported by Gaper's AI Agent Development services, keeping tracing decoupled from core business logic is essential.

Create logger.py to maintain the chain state in memory before writing to disk or database:

import json
import os
from audit_chain import AuditEntry

class AuditLogger:
    def __init__(self, log_file: str = "audit_log.json"):
        self.log_file = log_file
        self.last_hash = self._get_last_hash()

    def _get_last_hash(self) -> str:
        if not os.path.exists(self.log_file):
            return "0" * 64

        with open(self.log_file, "r") as f:
            lines = f.readlines()
            if not lines:
                return "0" * 64
            last_line = lines[-1]
            entry = json.loads(last_line)
            return entry["hash"]

    def log_action(self, agent_id: str, action: str, payload: dict) -> dict:
        entry = AuditEntry(
            agent_id=agent_id,
            action=action,
            payload=payload,
            previous_hash=self.last_hash
        )

        # Persist to append-only log file
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry.to_dict()) + "\n")

        self.last_hash = entry.hash
        return entry.to_dict()

Now, let's create a decorator to wrap agent functions automatically:

from functools import wraps

audit_logger = AuditLogger()

def trace_agent_action(agent_id: str, action_name: str):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            payload = {
                "inputs": {"args": args, "kwargs": kwargs}
            }
            try:
                result = func(*args, **kwargs)
                payload["output"] = result
                payload["status"] = "SUCCESS"
                audit_logger.log_action(agent_id, action_name, payload)
                return result
            except Exception as e:
                payload["error"] = str(e)
                payload["status"] = "FAILED"
                audit_logger.log_action(agent_id, action_name, payload)
                raise e
        return wrapper
    return decorator

Step 3: Execute Agent Actions and Log Events

Now let's simulate an AI agent performing actions. This could be a financial bot processing an invoice or an automated assistant operating in regulated fields, where AI in accounting and finance requires complete transparency and verifiable audit logs.

Create agent.py:

from logger import trace_agent_action

AGENT_ID = "finance_agent_v1"

@trace_agent_action(agent_id=AGENT_ID, action_name="fetch_user_balance")
def fetch_user_balance(user_id: str) -> float:
    # Simulated API call
    return 1450.50

@trace_agent_action(agent_id=AGENT_ID, action_name="execute_transfer")
def execute_transfer(user_id: str, target_account: str, amount: float) -> str:
    balance = fetch_user_balance(user_id)
    if amount > balance:
        raise ValueError("Insufficient funds")

    # Simulated execution
    return f"Transferred ${amount} to {target_account}"

if __name__ == "__main__":
    print("Running agent tasks...")
    execute_transfer("usr_99", "acc_441", 200.00)

    try:
        execute_transfer("usr_99", "acc_441", 5000.00)
    except Exception as e:
        print(f"Handled expected error: {e}")

Run python agent.py. This will generate audit_log.json containing JSON Lines data where each entry points to the hash of the entry before it.

Step 4: Verify Audit Log Integrity

The primary benefit of a cryptographic chain is simple verification. If an attacker or corrupt process alters any byte in audit_log.json, the hashes will fail to match downstream.

Create a verification script verify_chain.py:

import json
import hashlib

def verify_audit_log(file_path: str) -> bool:
    if not os.path.exists(file_path):
        print("No audit log found.")
        return False

    with open(file_path, "r") as f:
        lines = f.readlines()

    expected_previous_hash = "0" * 64

    for i, line in enumerate(lines):
        entry = json.loads(line)

        # 1. Check if previous_hash matches expected hash from previous iteration
        if entry["previous_hash"] != expected_previous_hash:
            print(f" Integrity failure at line {i+1}: Previous hash mismatch!")
            return False

        # 2. Recalculate hash of current entry
        data_to_hash = {
            "timestamp": entry["timestamp"],
            "agent_id": entry["agent_id"],
            "action": entry["action"],
            "payload": entry["payload"],
            "previous_hash": entry["previous_hash"]
        }
        recalculated_hash = hashlib.sha256(
            json.dumps(data_to_hash, sort_keys=True).encode('utf-8')
        ).hexdigest()

        if recalculated_hash != entry["hash"]:
            print(f" Integrity failure at line {i+1}: Hash content recalculation failed!")
            return False

        expected_previous_hash = entry["hash"]

    print(" Audit log integrity verified. No tampering detected.")
    return True

if __name__ == "__main__":
    verify_audit_log("audit_log.json")

To test verification, run python verify_chain.py. You will see a success message. If you manually open audit_log.json and change a dollar amount or timestamp, re-running the script will instantly report a chain break.

Hardening Storage for Production

Writing to a local JSON file works for prototyping, but production observability systems demand stricter protections:

AWS S3 Object Lock: Sync log batches to S3 buckets configured with Write-Once-Read-Many (WORM) storage in Compliance Mode. This prevents root users from deleting or altering log objects for a defined retention period.
Database Triggers: If using PostgreSQL, write log entries to an append-only table. You can revoke UPDATE and DELETE permissions for the application DB user, allowing only INSERT privileges.
Remote Log Aggregation: Ship logs off the host machine instantly via vector agents or streaming queues like Kafka to reduce local surface area for log tampering.

Summary

Building an immutable pipeline gives you immediate visibility into agent execution, making debugging straightforward while satisfying strict audit logs mandates. Cryptographic chaining guarantees that your execution history remains trustworthy, even when scaling complex multi-agent workflows.

If you need help building custom AI solutions, scaling agent infrastructure, or sourcing seasoned engineers to build compliant pipelines, check out Gaper for expert technical support and engineering talent.