引入框架并利用公司内部数据构建 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% 的费用。

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

1. 准备 Redis Vector Search 后端：启动一个 Redis 实例，用于将输入的 Prompt 转换为嵌入向量并进行存储。
2. 相似度判定与缓存返回：当新进入的 Prompt 向量与已有向量之间的余弦相似度达到 0.92 以上时，无需调用外部 LLM，直接返回已存储的响应。
3. 缓存未命中与保存：仅当相似度低于 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)])

`

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

1. 编写评估数据集：整理核心业务领域的查询与标准答案集，并设定允许的基准分数（Threshold）。
2. 结合 GitHub Actions 工作流：配置 Actions，使得在修改代码后提交 PR 时，自动运行 deepeval test run 命令。
3. 应用影子测试（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 模块并作为企业内部系统持续运营，需要建立周期的检查机制。

如果正在犹豫是选择自建模型服务（基于 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 天为一个阶段推进。

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