Call Loops and Token Cost Control Strategies for Custom AI Agents Beyond the Demo

Building an AI agent with company data using existing frameworks goes smoothly up to the prototype phase. The real problem starts the moment you push this code to production. An agent that worked fine on screen gets trapped by a single unexpected input, repeating calls endlessly and resulting in millions of won in API charges over the weekend.

A significant portion of field failures stems from state management failures and the absence of perimeter controls rather than the limitations of the LLM model itself. You may have built custom tools at the instruction of management, but you need to put an end to the situation where you cannot focus on your primary job because you are spending all day debugging unexpected agent hallucinations and handling a flood of token receipts.

Three Obstacles Encountered in Production Environments

Agents stepping out of the demo environment bump into three walls: black-box operations, non-deterministic loops, and data leakage risks.

If the inputs and outputs at each step—such as Chain-of-Thought reasoning, external tool calls, and vector database searches—are not logged, you cannot find the cause when an issue occurs. You end up wasting an entire day just tracing where the prompt got tangled.

Even more severe is the infinite loop. Agents with a ReAct (Reasoning + Acting) architecture keep resending the exact same request if the tool execution result is ambiguous.

┌─────────────────────────────────────────────────────────────────────────┐ │ ReAct Architecture Loop │ │ │ │ ┌────────────┐ User Query ┌────────────┐ Tool Call Request │ │ │ User │ ──────────────> │ Main LLM │ ────────────────────┐ │ │ └────────────┘ └────────────┘ │ │ │ ▲ ▼ │ │ │ Observation ┌──────┐ │ │ │ (Ambiguous/Failed) │ Tool │ │ │ └─────────────────────── │ A │ │ │ └──────┘ │ │ * Problem: When Observation fails, LLM retries Tool A endlessly. │ └─────────────────────────────────────────────────────────────────────────┘

When it fails to break out of the exit condition and continuously hits the API with identical arguments, a month's worth of token budget can be exhausted in just a few minutes. Cascading failures where primary and sub-agents repeatedly invoke each other account for more than 30% of overall system outages.

As conversation history accumulates and fills up the context window, model performance itself degrades. On top of that, prompts hardcoded as strings inside Python code force you to rebuild and redeploy the entire system just to modify a single phrase.

An Observability Framework to Sever Infinite Loops and Cost Bombs

You must separate local debugging solutions from production observability tools to establish distinct collection points. During the development phase, check RAG embedding quality and tool calls using Arize Phoenix based on OpenTelemetry. In a production environment, attach Langfuse with ClickHouse as the backend to monitor trace logs and token consumption in real time.

┌─────────────────────────────────────────────────────────────────────────┐ │ Semantic Caching & Routing Flow │ │ │ │ Client Query │ │ │ │ │ ▼ │ │ ┌───────────┐ Similarity >= 0.92? ┌─────────────────────────────┐ │ │ │ Redis Vector │ ─────────────────────────> │ Return Cached Response │ │ │ │ Cache │ (Cache Hit) │ (Latency -88%, Cost -86%) │ │ │ └───────────┘ └─────────────────────────────┘ │ │ │ │ │ │ (Cache Miss) │ │ ▼ │ │ ┌───────────┐ Execution & Save Cache ┌─────────────────────────────┐ │ │ │ External │ ────────────────────────> │ Store Result as Vector in │ │ │ │ LLM API │ │ Backend Database │ │ │ └───────────┘ └─────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────┘

API costs arising from repetitive identical queries can be blocked with a Semantic Caching layer.

According to analysis data of 60,000 queries released by AWS, establishing a semantic cache with cosine similarity criteria reduced LLM inference costs by up to 86% and improved response latency by 88%. LMSYS's RouteLLM framework also saved 85% in costs by splitting requests between high-cost and low-cost models depending on query difficulty.

Semantic caching is built in the following order:

Prepare a Redis Vector Search backend: Launch a Redis instance to store input prompts converted into embedding vectors.
Determine similarity and return cache: If the cosine similarity between a newly incoming prompt vector and an existing stored vector is 0.92 or higher, immediately retrieve and return the stored response without calling the external LLM.
Store cache misses: Call the external API only when the similarity falls below 0.92, and update the received response into Redis as new.

To fundamentally block infinite loops, a circuit breaker pattern that validates agent state must be injected directly into the pipeline control flow. Relying on default retry options provided by frameworks will cause exceptions to detonate, killing the entire service. It is safe to hash the tool name and argument values using SHA-256, store them in a list, and forcibly route to a different node if the same hash accumulates 3 consecutive times.

`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]

def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()

def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. If the total execution count exceeds 5, move to the designated fallback node
if state["steps"] > 5:
return "fallback_graceful_node"

# 2. Block immediately if the exact same Tool and Argument are called 3 consecutive times
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
    return "fallback_graceful_node"
    
# 3. Check for the normal completion keyword
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
    return "end"
    
return "continue_tools"

Regardless of the LLM's non-deterministic output state, this router validates conditional logic within the Python execution environment, physically preventing loop entry caused by hallucinations.

Prompt Version Decoupling and CI/CD Automation

Leaving prompts embedded inside code makes it impossible to catch regression failures where previously working features break after a model update. Prompts should be extracted out of Python code and managed as independent YAML files.

`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:

role: "system"
content: |
You are an internal database analysis agent.
You must respond according to the specified JSON schema, and generate follow-up questions if information is insufficient.
role: "user"
content: "{{ user_input }}"
`

Prompts decoupled in this manner are automatically tested in the CI/CD pipeline by pairing Pytest with DeepEval, an open-source evaluation framework. Set G-Eval metrics as a baseline to validate model response metrics before deployment.

`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams

1. Define evaluation criteria

correctness_metric = GEval(
name="Accuracy and Schema Compliance",
criteria="Does the LLM response accurately answer the question and fully adhere to the requested JSON format?",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)

2. Pytest-based regression test

@pytest.mark.parametrize(
"user_input, expected_output",
[
("2024년 1분기 매출 데이터를 요약해줘.", "1분기 총 매출은 50억 원입니다."),
("퇴직금 계산 규정을 알려줘.", "퇴직금은 근속연수 1년에 대해 30일분 이상의 평균임금입니다.")
]
)
def test_agent_regression(user_input, expected_output):
actual_output = run_in_house_agent(user_input)

test_case = LLMTestCase(
    input=user_input,
    actual_output=actual_output,
    expected_output=expected_output
)

# Abort build if below the configured baseline score
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])

The evaluation framework integrates across three stages:

Author evaluation datasets: Create sets of key domain queries and ground truths, and define acceptable threshold scores.
Integrate GitHub Actions workflows: Configure actions so that the deepeval test run command executes automatically when opening a PR after code modifications.
Apply shadow testing: Asynchronously duplicate and transmit real traffic requests before merging into the main branch to compare response quality and latency against the previous version prior to final deployment.

`yaml
name: AI Agent Evaluation Gate

on:
pull_request:
branches: [ main ]

jobs:
eval-gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Dependencies
run: |
pip install poetry
poetry install
- name: Run DeepEval Suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
poetry run deepeval test run tests/test_evals.py
`

Applying Masking and Access Control

To prevent internal data leakage, guardrails that anonymize Personally Identifiable Information (PII) before API requests leave the network are essential. Attaching the Microsoft Presidio engine at the gateway will automatically mask Resident Registration Numbers (RRN), emails, phone numbers, and employee IDs.

`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

1. Register Analyzer Engine and employee ID rules

analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)

2. Initialize Anonymizer Engine

anonymizer = AnonymizerEngine()

def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)

anonymized_result = anonymizer.anonymize(
    text=raw_prompt,
    analyzer_results=results,
    operators={
        "DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
        "EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
    }
)
return anonymized_result.text

When attaching RAG retrieval, measures must also be taken so that only documents matching the user's authorization level are pulled. Vector databases like Qdrant or Pinecone handle permission isolation via Metadata Filtering during similarity searches.

`python
from qdrant_client import QdrantClient
from qdrant_client.http import models

client = QdrantClient(host="localhost", port=6333)

def search_documents_with_rbac(query_vector: list, user_department: str, user_clearance_level: int):
search_result = client.search(
collection_name="enterprise_knowledge_base",
query_vector=query_vector,
query_filter=models.Filter(
must=[
models.FieldCondition(
key="department",
match=models.MatchValue(value=user_department)
),
models.FieldCondition(
key="security_level",
range=models.Range(lte=user_clearance_level)
)
]
),
limit=5
)
return search_result
`

The fundamental defense against prompt injection is completely isolating the system prompt channel from the user input channel. Risky tasks that modify databases or call external APIs must be bound to human approval workflows (Human-in-the-Loop) or restricted to run within isolated sandboxes for safety.

Infrastructure Lifecycle and Stage-by-Stage Alignment

Replacing SaaS modules to continuously operate as in-house systems requires periodic inspection routines.

Cycle	Operational Checklist Item	Detailed Tasks
Daily	Error rate & token usage	Check HTTP 5xx error rates and department token consumption on the Langfuse dashboard
Daily	Circuit breaker trip logs	Collect tool names and argument patterns from sessions severed by loop detection, then patch them
Weekly	Extract RAG search failure queries	Filter requests with similarity scores below 0.6 and add them to the test ground truth set
Weekly	Verify PII masking false positives	Sample Presidio processing logs to verify if any masking was missed
Monthly	Update CI/CD evaluation criteria	Modify automated test cases to reflect business changes

If you are weighing self-hosted model serving (vLLM-based) against external API subscriptions, calculate based on daily traffic.

The monthly cost for an AWS EC2 g5.2xlarge instance equipped with 1 NVIDIA A10G GPU is approximately $880. On the other hand, GPT-4o API pricing is around $2.50 per 1 million input tokens and $10.00 per 1 million output tokens. If daily traffic is under 50 million tokens, an API subscription approach is advantageous considering DevOps engineer maintenance overhead and fixed GPU costs. Transitioning to self-hosted vLLM serving makes sense only when daily traffic exceeds 100 million tokens or complete internal network isolation is mandatory.

To prepare for main model failures, construct a 4-tier graceful degradation routing architecture.

┌─────────────────────────────────────────────────────────────────────────┐ │ 4-Tier Graceful Degradation Architecture │ │ │ │ [Tier 1] Primary High-Performance Model (e.g., GPT-4o) │ │ │ │ │ ▼ (API Failure / Timeout / Circuit Breaker) │ │ [Tier 2] Lightweight Routing Model (e.g., GPT-4o-mini / On-Prem vLLM) │ │ │ │ │ ▼ (Continuous Outage) │ │ [Tier 3] Deterministic Regex & SQL Rule Engine │ │ │ │ │ ▼ (Unrecoverable Error) │ │ [Tier 4] Static Error Message & Async Admin Ticket Generation │ └─────────────────────────────────────────────────────────────────────────┘

Operational Stabilization Roadmap

Refining custom agent systems should be carried out in 90-day increments.

Period	Implementation Goal	Concrete Execution Details
Days 1–30	Visibility & Data Protection	Install Presidio masking engine and integrate Langfuse logging SDK
Days 31–60	Loop Interruption & Cost Reduction	Apply Redis semantic caching and connect Python-based circuit breaker
Days 61–90	Automated Verification & Access Control	Manage prompts in Git, integrate DeepEval CI/CD, apply Qdrant RBAC metadata

In the first month, establish observability solutions to prevent PII leaks and ensure every request is logged. In the second month, attach repetitive prompt caching and loop-blocking routers to cut off unexpected cost expenditures. In the final month, completing prompt version control and automated test evaluation frameworks enables stable operation without the fear of agent malfunctions every time you deploy.

Call Loops and Token Cost Control Strategies for Custom AI Agents Beyond the Demo

Three Obstacles Encountered in Production Environments

Agents stepping out of the demo environment bump into three walls: black-box operations, non-deterministic loops, and data leakage risks.

Even more severe is the infinite loop. Agents with a ReAct (Reasoning + Acting) architecture keep resending the exact same request if the tool execution result is ambiguous.

An Observability Framework to Sever Infinite Loops and Cost Bombs

API costs arising from repetitive identical queries can be blocked with a Semantic Caching layer.

Semantic caching is built in the following order:

Prepare a Redis Vector Search backend: Launch a Redis instance to store input prompts converted into embedding vectors.
Determine similarity and return cache: If the cosine similarity between a newly incoming prompt vector and an existing stored vector is 0.92 or higher, immediately retrieve and return the stored response without calling the external LLM.
Store cache misses: Call the external API only when the similarity falls below 0.92, and update the received response into Redis as new.

`python
import hashlib
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
steps: int
tool_hashes: List[str]

def hash_tool_call(tool_name: str, tool_args: str) -> str:
raw_str = f"{tool_name}:{tool_args}"
return hashlib.sha256(raw_str.encode('utf-8')).hexdigest()

def agent_circuit_breaker_router(state: AgentState) -> str:
# 1. If the total execution count exceeds 5, move to the designated fallback node
if state["steps"] > 5:
return "fallback_graceful_node"

# 2. Block immediately if the exact same Tool and Argument are called 3 consecutive times
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
    return "fallback_graceful_node"
    
# 3. Check for the normal completion keyword
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
    return "end"
    
return "continue_tools"

Regardless of the LLM's non-deterministic output state, this router validates conditional logic within the Python execution environment, physically preventing loop entry caused by hallucinations.

Prompt Version Decoupling and CI/CD Automation

`yaml
name: "agent_reasoning"
version: "1.2.0"
model: "gpt-4o"
temperature: 0.1
messages:

role: "system"
content: |
You are an internal database analysis agent.
You must respond according to the specified JSON schema, and generate follow-up questions if information is insufficient.
role: "user"
content: "{{ user_input }}"
`

`python
import pytest
from deepeval import assert_test
from deepeval.metrics import GEval, TaskCompletionMetric
from deepeval.test_case import LLMTestCase, SingleTurnParams

1. Define evaluation criteria

2. Pytest-based regression test

test_case = LLMTestCase(
    input=user_input,
    actual_output=actual_output,
    expected_output=expected_output
)

# Abort build if below the configured baseline score
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])

The evaluation framework integrates across three stages:

Author evaluation datasets: Create sets of key domain queries and ground truths, and define acceptable threshold scores.
Integrate GitHub Actions workflows: Configure actions so that the deepeval test run command executes automatically when opening a PR after code modifications.
Apply shadow testing: Asynchronously duplicate and transmit real traffic requests before merging into the main branch to compare response quality and latency against the previous version prior to final deployment.

`yaml
name: AI Agent Evaluation Gate

on:
pull_request:
branches: [ main ]

Applying Masking and Access Control

`python
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

1. Register Analyzer Engine and employee ID rules

analyzer = AnalyzerEngine()
employee_id_recognizer = PatternRecognizer(
supported_entity="EMPLOYEE_ID",
regex="EMP-[0-9]{6}",
score=0.95
)
analyzer.registry.add_recognizer(employee_id_recognizer)

2. Initialize Anonymizer Engine

anonymizer = AnonymizerEngine()

def sanitize_user_prompt(raw_prompt: str) -> str:
results = analyzer.analyze(
text=raw_prompt,
entities=["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "EMPLOYEE_ID"],
language="en"
)

anonymized_result = anonymizer.anonymize(
    text=raw_prompt,
    analyzer_results=results,
    operators={
        "DEFAULT": OperatorConfig("replace", {"new_value": "<REDACTED>"}),
        "EMPLOYEE_ID": OperatorConfig("mask", {"chars_to_mask": 6, "masking_char": "*", "from_end": True})
    }
)
return anonymized_result.text

`python
from qdrant_client import QdrantClient
from qdrant_client.http import models

client = QdrantClient(host="localhost", port=6333)

Infrastructure Lifecycle and Stage-by-Stage Alignment

Replacing SaaS modules to continuously operate as in-house systems requires periodic inspection routines.

Cycle	Operational Checklist Item	Detailed Tasks
Daily	Error rate & token usage	Check HTTP 5xx error rates and department token consumption on the Langfuse dashboard
Daily	Circuit breaker trip logs	Collect tool names and argument patterns from sessions severed by loop detection, then patch them
Weekly	Extract RAG search failure queries	Filter requests with similarity scores below 0.6 and add them to the test ground truth set
Weekly	Verify PII masking false positives	Sample Presidio processing logs to verify if any masking was missed
Monthly	Update CI/CD evaluation criteria	Modify automated test cases to reflect business changes

If you are weighing self-hosted model serving (vLLM-based) against external API subscriptions, calculate based on daily traffic.

To prepare for main model failures, construct a 4-tier graceful degradation routing architecture.

Operational Stabilization Roadmap

Refining custom agent systems should be carried out in 90-day increments.

Period	Implementation Goal	Concrete Execution Details
Days 1–30	Visibility & Data Protection	Install Presidio masking engine and integrate Langfuse logging SDK
Days 31–60	Loop Interruption & Cost Reduction	Apply Redis semantic caching and connect Python-based circuit breaker
Days 61–90	Automated Verification & Access Control	Manage prompts in Git, integrate DeepEval CI/CD, apply Qdrant RBAC metadata

Call Loops and Token Cost Control Strategies for Custom AI Agents Beyond the Demo

Related Video

"Build vs Buy" is shifting towards "build"

Call Loops and Token Cost Control Strategies for Custom AI Agents Beyond the Demo

Three Obstacles Encountered in Production Environments

An Observability Framework to Sever Infinite Loops and Cost Bombs

Prompt Version Decoupling and CI/CD Automation

1. Define evaluation criteria

2. Pytest-based regression test

Applying Masking and Access Control

1. Register Analyzer Engine and employee ID rules

2. Initialize Anonymizer Engine

Infrastructure Lifecycle and Stage-by-Stage Alignment

Operational Stabilization Roadmap

Comments (0)

Call Loops and Token Cost Control Strategies for Custom AI Agents Beyond the Demo

Three Obstacles Encountered in Production Environments

An Observability Framework to Sever Infinite Loops and Cost Bombs

Prompt Version Decoupling and CI/CD Automation

1. Define evaluation criteria

2. Pytest-based regression test

Applying Masking and Access Control

1. Register Analyzer Engine and employee ID rules

2. Initialize Anonymizer Engine

Infrastructure Lifecycle and Stage-by-Stage Alignment

Operational Stabilization Roadmap