演示结束后的自定义 AI Agent 调用循环与 Token 成本控制方案

引入框架并利用公司内部数据构建 AI Agent，在原型阶段通常十分顺畅。真正的问题始于将这段代码部署到生产环境的那一刻。在演示界面中表现良好的 Agent，可能会因为一个意料之外的输入而陷入死循环，导致 API 调用无休止地重复，甚至在周末短短几天内产生高达数万元的 API 费用。

现场发生的绝大多数故障，其根本原因往往不在于 LLM 模型本身的局限，而在于状态管理的失败以及缺乏外围控制。为了完成管理层的指示，我们开发了自定义工具，但绝不能让团队整天陷于调试无预警爆发的 Agent 幻觉、处理砸过来的 Token 账单中，从而无法顾及本职工作。

生产环境中面临的三大障碍

脱离演示环境后，Agent 会撞上三座大山：黑盒运行、非确定性循环以及数据泄露风险。

如果推理（Chain-of-Thought）过程、外部工具调用（Tool Call）以及向量数据库检索等各个阶段的输入输出没有记录在日志中，一旦出现问题将无法追踪原因。光是排查 Prompt 在哪里出问题，就可能耗费一整天的时间。

更严重的是无限循环。采用 ReAct（Reasoning + Acting）架构的 Agent，在工具执行结果模糊不清时，会不断重复发送相同的请求。

┌─────────────────────────────────────────────────────────────────────────┐ │ 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. │ └─────────────────────────────────────────────────────────────────────────┘

如果无法触发退出条件，并持续以相同的参数轰炸 API，只需短短几分钟，整月的 Token 预算就会耗尽。主 Agent 与子 Agent 相互调用并不断重试所引发的连锁失败，占到了整个系统故障原因的 30% 以上。

此外，对话历史不断累积会导致上下文窗口（Context Window）填满，进而引发模型本身的性能下降。再加上硬编码在 Python 代码中的字符串 Prompt，哪怕只是修改一句简单的文案，也需要重新构建并部署整个系统。

切断无限循环与成本爆炸的可观测体系

必须将本地调试解决方案与生产环境的可观测工具相分离，以精准抓取数据采集点。在开发阶段，使用基于 OpenTelemetry 的 Arize Phoenix 来验证 RAG 嵌入质量和工具调用；在生产环境中，则接入以 ClickHouse 为后端存储的 Langfuse，对 Trace 日志和 Token 消耗量进行实时监控。

┌─────────────────────────────────────────────────────────────────────────┐ │ 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 费用，可以通过语义缓存（Semantic Caching）层来进行拦截。

根据 AWS 发布的 60,000 条查询分析数据，当引入设置了余弦相似度标准的语义缓存后，LLM 推理成本最高降低了 86%，响应延迟也改善了 88%。LMSYS 的 RouteLLM 框架同样根据问题难度将请求分流至高成本与低成本模型，从而节省了 85% 的费用。

构建语义缓存的步骤如下：

准备 Redis Vector Search 后端：启动一个 Redis 实例，用于将输入的 Prompt 转换为嵌入向量并进行存储。
相似度判定与缓存返回：当新进入的 Prompt 向量与已有向量之间的余弦相似度达到 0.92 以上时，无需调用外部 LLM，直接返回已存储的响应。
缓存未命中与保存：仅当相似度低于 0.92 时才调用外部 API，并将接收到的响应更新保存至 Redis 中。

要从根本上阻断无限循环，必须将用于验证 Agent 状态的熔断器（Circuit Breaker）模式直接嵌入到管道控制流中。如果单纯依赖框架提供的默认重试选项，一旦抛出异常，整个服务都会崩溃。安全做法是将工具名称和参数值通过 SHA-256 进行哈希处理并存入列表，当相同的哈希值连续出现 3 次时，强制跳转至其他节点。

`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. 总执行次数超过 5 次时，跳转至预设的退路节点
if state["steps"] > 5:
return "fallback_graceful_node"

# 2. 连续 3 次调用相同的 Tool 和 Argument 时立即拦截
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
    return "fallback_graceful_node"
    
# 3. 检查正常结束关键字
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
    return "end"
    
return "continue_tools"

无论 LLM 的输出状态是否具备确定性，由于该路由器是在 Python 运行环境中进行条件判断，因此能够物理性地阻止因幻觉而陷入的死循环。

Prompt 版本隔离与 CI/CD 自动化

如果将 Prompt 随意搁置在代码内部，一旦修改模型，就无法捕获导致先前正常功能失效的回归故障（Regression Failure）。应当将 Prompt 剥离出 Python 代码，作为独立的 YAML 文件进行管理。

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

role: "system"
content: |
คุณคือ AI Agent 分析公司内部数据库的助手。
必须严格按照指定的 JSON Schema 进行回答，如果信息不足，请生成补充提问。
role: "user"
content: "{{ user_input }}"
`

将分离出来的 Prompt 与开源评估框架 DeepEval 以及 Pytest 结合，即可在 CI/CD 管道中实现自动化测试。以 G-Eval 指标作为基准线，在部署前对模型的响应数值进行验证。

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

1. 定义评估标准

correctness_metric = GEval(
name="准确性及 Schema 遵循度",
criteria="LLM 响应是否准确回答了问题，并且完美遵循了要求的 JSON 格式？",
evaluation_params=[SingleTurnParams.ACTUAL_OUTPUT, SingleTurnParams.EXPECTED_OUTPUT],
threshold=0.7
)

2. 基于 Pytest 的回归测试

@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
)

# 未达到设定的基准分数时中断构建
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])

评估体系分三个阶段集成：

编写评估数据集：整理核心业务领域的查询与标准答案集，并设定允许的基准分数（Threshold）。
结合 GitHub Actions 工作流：配置 Actions，使得在修改代码后提交 PR 时，自动运行 deepeval test run 命令。
应用影子测试（Shadow Testing）：在合并至主分支前，异步复制生产环境的真实流量请求进行发送，对比新旧版本的响应质量与延迟时间，确认无误后再进行最终部署。

`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
`

实施脱敏与权限控制

为防止公司内部数据泄露，在 API 请求发送至外部之前，需要设置安全护栏对个人身份信息（PII）进行去标识化。在网关层挂载 Microsoft Presidio 引擎，即可自动对身份证号、电子邮件、电话号码以及员工编号进行掩码处理。

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

1. 注册 Analyzer Engine 及工号规则

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. 初始化 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

在接入 RAG 检索时，也必须采取措施，确保仅拉取符合用户权限等级的文档。Qdrant 或 Pinecone 等向量数据库在进行相似度检索时，可以通过元数据过滤（Metadata Filtering）来处理权限划分。

`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
`

防御 Prompt 注入的基础在于将系统提示词通道与用户输入通道完全隔离。对于修改数据库或调用外部 API 等高风险操作，引入人工审批流程（Human-in-the-Loop），或者将其执行权限限制在隔离的沙盒内部，才能保证系统安全。

基础设施生命周期与分阶段一致性保障

要替代 SaaS 模块并作为企业内部系统持续运营，需要建立周期的检查机制。

周期	运维检查项目	详细工作
每日	错误率及 Token 使用量	在 Langfuse 仪表板中检查 HTTP 5xx 错误率及各部门 Token 消耗量
每日	熔断器拦截记录	收集因检测到死循环而中断的会话的工具名称和参数模式并进行修复
每周	提取 RAG 检索失败查询	筛选相似度得分低于 0.6 的请求，将其补充至测试标准答案集中
每周	验证个人信息脱敏误判	抽样检查 Presidio 处理日志，确认是否存在脱敏遗漏
每月	更新 CI/CD 评估基准	根据业务变更情况，修改自动化测试用例

如果正在犹豫是选择自建模型服务（基于 vLLM）还是订阅外部 API，可以根据日流量来进行估算。

包含 1 张 NVIDIA A10G GPU 的 AWS EC2 g5.2xlarge 实例月租费用约为 $880。相比之下，GPT-4o API 的单价大约为输入每 100 万 Token $2.50，输出每 100 万 Token $10.00。如果日流量低于 5,000 万 Token，考虑到 DevOps 工程师的运维成本和 GPU 固定成本，采用 API 订阅方式更为划算。只有当日流量突破 1 亿 Token，或者必须实现绝对的企业内网隔离时，转向基于 vLLM 的自建服务才是合理的选择。

为了应对主模型故障，可以构建 4 级降级路由架构：

┌─────────────────────────────────────────────────────────────────────────┐ │ 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 │ └─────────────────────────────────────────────────────────────────────────┘

运维稳定化顺序

自定义 Agent 系统的优化迭代工作建议以 90 天为一个阶段推进。

期间	实施目标	具体工作内容
第 1~30 天	可视化与数据保护	安装 Presidio 脱敏引擎并植入 Langfuse 日志 SDK
第 31~60 天	阻断循环与控制成本	应用 Redis 语义缓存并接入基于 Python 的熔断器
第 61~90 天	自动化验证与权限控制	实现 Prompt 的 Git 管理、DeepEval CI/CD 联动以及 Qdrant RBAC 元数据应用

首月建立可观测性解决方案，防止个人信息泄露并确保所有请求记录日志。次月接入重复 Prompt 缓存与循环阻断路由器，切断意料之外的成本支出。最后一个月完善 Prompt 版本管理与自动化测试评估体系，即可避免每次部署时对 Agent 误动作的担忧，实现稳定运行。

演示结束后的自定义 AI Agent 调用循环与 Token 成本控制方案

生产环境中面临的三大障碍

脱离演示环境后，Agent 会撞上三座大山：黑盒运行、非确定性循环以及数据泄露风险。

更严重的是无限循环。采用 ReAct（Reasoning + Acting）架构的 Agent，在工具执行结果模糊不清时，会不断重复发送相同的请求。

切断无限循环与成本爆炸的可观测体系

对于因重复查询而产生的 API 费用，可以通过语义缓存（Semantic Caching）层来进行拦截。

构建语义缓存的步骤如下：

准备 Redis Vector Search 后端：启动一个 Redis 实例，用于将输入的 Prompt 转换为嵌入向量并进行存储。
相似度判定与缓存返回：当新进入的 Prompt 向量与已有向量之间的余弦相似度达到 0.92 以上时，无需调用外部 LLM，直接返回已存储的响应。
缓存未命中与保存：仅当相似度低于 0.92 时才调用外部 API，并将接收到的响应更新保存至 Redis 中。

`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. 总执行次数超过 5 次时，跳转至预设的退路节点
if state["steps"] > 5:
return "fallback_graceful_node"

# 2. 连续 3 次调用相同的 Tool 和 Argument 时立即拦截
hashes = state.get("tool_hashes", [])
if len(hashes) >= 3 and hashes[-1] == hashes[-2] == hashes[-3]:
    return "fallback_graceful_node"
    
# 3. 检查正常结束关键字
last_message = state["messages"][-1]
if "FINAL_ANSWER" in last_message.content:
    return "end"
    
return "continue_tools"

无论 LLM 的输出状态是否具备确定性，由于该路由器是在 Python 运行环境中进行条件判断，因此能够物理性地阻止因幻觉而陷入的死循环。

Prompt 版本隔离与 CI/CD 自动化

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

role: "system"
content: |
คุณคือ AI Agent 分析公司内部数据库的助手。
必须严格按照指定的 JSON Schema 进行回答，如果信息不足，请生成补充提问。
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. 定义评估标准

2. 基于 Pytest 的回归测试

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

# 未达到设定的基准分数时中断构建
assert_test(test_case, [correctness_metric, TaskCompletionMetric(threshold=0.8)])

评估体系分三个阶段集成：

编写评估数据集：整理核心业务领域的查询与标准答案集，并设定允许的基准分数（Threshold）。
结合 GitHub Actions 工作流：配置 Actions，使得在修改代码后提交 PR 时，自动运行 deepeval test run 命令。
应用影子测试（Shadow Testing）：在合并至主分支前，异步复制生产环境的真实流量请求进行发送，对比新旧版本的响应质量与延迟时间，确认无误后再进行最终部署。

`yaml
name: AI Agent Evaluation Gate

on:
pull_request:
branches: [ main ]

实施脱敏与权限控制

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

1. 注册 Analyzer Engine 及工号规则

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. 初始化 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)

基础设施生命周期与分阶段一致性保障

要替代 SaaS 模块并作为企业内部系统持续运营，需要建立周期的检查机制。

周期	运维检查项目	详细工作
每日	错误率及 Token 使用量	在 Langfuse 仪表板中检查 HTTP 5xx 错误率及各部门 Token 消耗量
每日	熔断器拦截记录	收集因检测到死循环而中断的会话的工具名称和参数模式并进行修复
每周	提取 RAG 检索失败查询	筛选相似度得分低于 0.6 的请求，将其补充至测试标准答案集中
每周	验证个人信息脱敏误判	抽样检查 Presidio 处理日志，确认是否存在脱敏遗漏
每月	更新 CI/CD 评估基准	根据业务变更情况，修改自动化测试用例

如果正在犹豫是选择自建模型服务（基于 vLLM）还是订阅外部 API，可以根据日流量来进行估算。

为了应对主模型故障，可以构建 4 级降级路由架构：

运维稳定化顺序

自定义 Agent 系统的优化迭代工作建议以 90 天为一个阶段推进。

期间	实施目标	具体工作内容
第 1~30 天	可视化与数据保护	安装 Presidio 脱敏引擎并植入 Langfuse 日志 SDK
第 31~60 天	阻断循环与控制成本	应用 Redis 语义缓存并接入基于 Python 的熔断器
第 61~90 天	自动化验证与权限控制	实现 Prompt 的 Git 管理、DeepEval CI/CD 联动以及 Qdrant RBAC 元数据应用

演示结束后的自定义 AI Agent 调用循环与 Token 成本控制方案

Related Video

“自研还是外购”正向“自研”倾斜

演示结束后的自定义 AI Agent 调用循环与 Token 成本控制方案

生产环境中面临的三大障碍

切断无限循环与成本爆炸的可观测体系

Prompt 版本隔离与 CI/CD 自动化

1. 定义评估标准

2. 基于 Pytest 的回归测试

实施脱敏与权限控制

1. 注册 Analyzer Engine 及工号规则

2. 初始化 Anonymizer Engine

基础设施生命周期与分阶段一致性保障

运维稳定化顺序

Comments (0)

演示结束后的自定义 AI Agent 调用循环与 Token 成本控制方案

生产环境中面临的三大障碍

切断无限循环与成本爆炸的可观测体系

Prompt 版本隔离与 CI/CD 自动化

1. 定义评估标准

2. 基于 Pytest 的回归测试

实施脱敏与权限控制

1. 注册 Analyzer Engine 及工号规则

2. 初始化 Anonymizer Engine

基础设施生命周期与分阶段一致性保障

运维稳定化顺序