在2025年7月,一个基于Claude Code的递归循环在五小时内消耗了16,000美元到50,000美元不等的费用。并没有出现系统崩溃或错误,只是这些代理程序一直按照指令执行操作,因为没有人告诉它们何时应该停止。
四个月后,一个由四个代理程序组成的LangChain循环运行了11天,最终花费了47,000美元。直到账单寄到手中,才有人注意到这个问题。在测试阶段,这个系统运行正常,所有代理程序也都严格按照指令行事——同样的情况再次发生了。
本教程就是为了探讨这种“缺失的指令”问题而设计的。
你将构建五个简单的Python工具,这些工具能够在大多数代理循环出现故障之前及时发现并阻止问题的发生:
-
一个规范编写工具,它要求你在循环开始之前就明确定义“完成条件”。
-
一个断路器机制,当循环超出预设的限制时,它会立即终止该循环。
-
一个记录系统,它会将循环中的每一步操作都记录在只允许追加数据的SQLite审计日志中。
-
一个将这三个工具结合在一起的代理循环框架。
-
一个审核机制,它在任何下游系统接收数据之前,都会要求人工进行确认。
完成这些开发后,你将会得到一个可以直接应用于任何代理项目的功能完备的代码库。完整代码请访问github.com/dannwaneri/production-safe-agent-loop。
目录
为什么这种问题会不断发生
导致企业遇到麻烦的数学原理其实很简单:一个聊天机器人的每次交互成本大约为0.04美元,而一个由多个代理程序组成的复杂工作流程的成本则为1.20美元。这意味着两者之间的成本差距高达30倍——而在处理复杂任务时,这一差距甚至可能达到70倍。
问题并不在于代理程序本身价格昂贵,而在于大多数团队在预算规划时只考虑了聊天机器人的成本,却忽略了部署代理程序架构所需的费用。Gartner的研究发现,试点阶段的聊天机器人与正式生产环境中的代理程序在工作流程上的成本差距介于5倍到30倍之间。FinOps Foundation在2026年发布的报告中也指出,73%的企业表示,人工智能相关成本的实际情况超出了最初的预测。
一旦了解了这个机制,就会发现它其实非常简单。当一个代理在执行任务时失败并尝试重新运行时,它并不会从零开始执行,而是会重新读取所有的上下文信息——包括之前所有失败的尝试记录——然后再进行尝试。第一次迭代需要消耗100个令牌,第二次迭代需要200个令牌,而第十次迭代则可能需要数千个令牌。也就是说,每次失败都会让使用者付出相应的代价,而这些代价是以毫秒为单位计算的。
# 这整个流程可以用三行代码来表示:
while True:
result = agent.run(task)
# 什么时候才算完成呢?...
而恰恰就是这个“什么时候才算完成”这个问题,导致了大量的资源被消耗掉。
还有另一个让情况变得更糟的因素是:这些代理在遇到问题时并不会出现明显的错误提示。传统的代码在遇到未知状态时会崩溃,而大型语言模型在面对模糊性或不确定性时,会尝试继续执行并提供帮助。它们会重新尝试执行任务,会调整调用命令的格式,甚至会启动额外的验证程序。然而,由于没有人明确界定“正确”的标准是什么,这些尝试最终往往还是会导致问题再次发生。在各种监控仪表板上,这些操作看起来都很正常——活动量、调用次数、完成率等等——但实际上却在悄悄地消耗着你的预算。
根据Gartner的预测,到2027年,有40%的基于代理的技术项目会因为经济原因而被终止。而其中大部分失败其实都是可以避免的。关键不在于使用更好的模型,而在于设置合理的退出条件。
先决条件
-
Python 3.10或更高版本
-
Anthropic的API密钥(或其他服务提供商提供的密钥——具体细节稍后介绍)
-
对Python类和SQLite有一定的了解
git clone https://github.com/dannwaneri/production-safe-agent-loop
cd production-safe-agent-loop
pip install -r requirements.txt
export ANTHROPIC_API_KEY=sk-...
第1阶段:在开始开发之前先明确“完成的标准”
在代理程序的开发过程中,最代价高昂的错误并不是选择了错误的模型,也不是没有设置重试次数限制。真正的错误在于在还没有弄清楚“完成的标准”之前就开始了开发工作。
大多数团队都无法回答这个问题。这并不是因为他们粗心大意,而是因为在他们打开终端开始编写代码之前,没有任何东西迫使他们去思考这个问题。而“规范编写者”恰恰就是那个能够起到这种强制作用的角色。
# spec_writer.py
from spec_writer import SpecWriter
spec = SpecWriter(db_path="spec.db").run()
当你调用.run()这个方法时,它不会立即返回结果,直到你回答了以下三个问题为止:
-
这个代理程序的作用是什么?
-
这个代理程序不能做什么?
-
用一句话来描述“完成的标准”是什么?
-
input_hash存储的是输入字符串的SHA-256哈希值,而不是原始输入字符串本身。这样做有两个好处:首先,可以检测出不同运行环境中相同的输入数据;其次,个人身份信息永远不会被记录到审计日志中。 -
pass_fail被定义为整数类型,而不是布尔类型。这是因为SQLite数据库中没有布尔类型,使用1和0来表示通过或失败是符合Python的编程习惯的,同时也能确保数据库中的数据类型与SQL标准保持一致。 -
created_at字段存储的是当前时间戳,但这个时间戳是按照UTC时区进行计算的。在Python 3.12之后,datetime.utcnow()这个函数已经被弃用,因此使用datetime.now(timezone.utc).isoformat()才能确保时间戳的正确性。 -
circuit_breaker.check(turn_count, accumulated_tokens)—— 如果超过了任何限制,就会触发这个检查 -
client.messages.create(...)—— 实际的调用大型语言模型的操作 -
ledger.write(...)—— 将相关数据写入日志文件,仅支持追加操作 -
如果
stop_reason == "end_turn",则结束当前轮次;否则继续循环 -
while True:整个循环都被包裹在一个try/except CircuitBreakerError结构中。检查会在每一轮循环的开始阶段进行,因此无论是第1轮还是第6轮出现异常,都能被及时发现。 -
在每行账本数据中,都使用
input_str=task来获取原始任务内容,而不是最后的辅助信息。通过input_hash这一列,可以将整个运行过程中具有相同输入值的记录归为一组。 -
对于每一个返回
False结果的LLM循环,都会设置pass_fail=True;只有当循环出现异常时,这个标志才会被设置为False。这个通过/失败标志用于判断循环是否合法地完成了执行过程,而不是模型输出的结果是否正确。质量评估则是另一个需要考虑的问题。 -
_system_prompt()函数会使用所有的三个规范字段,而不仅仅是done_looks_like。模型同样需要了解“它不能做什么”这一方面的信息。 -
应该使用
time.perf_counter()而不是time.time(),因为前者是单调递增的,在运行过程中时钟时间的调整不会影响其计数值。 -
最初承诺的内容——这些信息来自规范文件,包括模型能做什么、不能做什么,以及完成任务后应该呈现什么样的结果。
-
验收标准——即
done_looks_like字段所定义的明确基准。 -
差异分析——包括第一轮输入数据与最终输出结果之间的对比、已完成的任务轮次数、总token数量,以及循环是否出现了异常。
-
证据资料——该会话期间所有的账本记录,包括每一轮的通过/失败情况、token数量的变化以及执行耗时等信息。
-
未解决的疑问——这些疑问是根据那些出现异常的记录和失败的循环环节总结出来的。如果所有数据都正常,这一栏就会显示为空。
其中第三个问题才是真正关键的问题,也是最难回答的。“这个代理程序会审核网站内容”这样的答案是不够的。而“这个代理程序会爬取目标网址,提取所有的和标签,检查是否存在缺失或过长的标签,然后停止执行”这样的答案才符合要求。只有明确了这样的标准,才能让系统知道应该如何进行验证。
编写好的规范会被保存到SQLite数据库中,并返回一个包含session_id字段的SpecResult数据对象。这个ID就是连接你的规范文件、数据库记录以及程序执行结果的纽带,它能够确保所有的操作都能被从头到尾追踪起来。
@dataclass(frozen=True)
class SpecResult:
what_it_does: str
what_it_does_not: str
done_looks_like: str
session_id: str
frozen=True这一设置非常重要。规范文件是一种具有约束力的文件,而不是草稿。一旦编写完成,后续的所有操作都必须严格遵循这些规范,不允许在运行过程中对其进行修改。
在测试过程中,SpecWriter会接受可插入的input_fn和output_fn》函数作为输入参数,因此无需对标准输入进行任何修改。具体的使用示例可以参考文件tests/test_spec_writer.py——该测试用例中使用了scripted_input辅助工具,该工具可以从生成器中获取答案,并通过pytest的tmp_path机制将结果写入每个测试对应的SQLite文件中。需要注意的是,使用SQLite的:memory:数据库模式并不安全,因为SpecWriter会为每种测试方法创建一个新的连接,而每个:memory:连接都代表一个独立的数据库环境。
阶段2:在运行时强制执行操作结束
在代码的上游明确规定退出条件是一种必要的纪律要求,而“断路器机制”则是确保这些规则得到严格执行的有效手段。
# circuit_breaker.py
from circuit_breaker import CircuitBreaker, CircuitBreakerError
breaker = CircuitBreaker(turn_limit=5, token_limit=15000)
breaker.check(turn_count, accumulated_tokens) # 当超过限制时会触发异常
这里设置了两个严格的上限限制:turn_limit规定了循环最多可以调用LLM多少次,而token_limit则限制了所有调用过程中消耗的总令牌数。一旦任何一个上限被突破,就会立即引发CircuitBreakerError异常。
这些限制是非常严格的:即使turn_count恰好等于turn_limit,也会触发异常;不允许有任何缓冲期或警告提示,系统会立即停止运行,迫使人类审核员介入处理。
from dataclasses import dataclass
@dataclass
class CircuitBreakerError(Exception):
reason: str # "turn_ceiling" 或 "token_ceiling"
turn_count: int
accumulated_tokens: int
def __post_init__(self) -> None:
super().__init__(
f"断路器触发异常:{self.reason} "
f"(当前调用次数:{self.turn_count}, 总消耗令牌数:{self.accumulated_tokens}")
)
class CircuitBreaker:
def __init__(self, turn_limit: int = 5, token_limit: int = 15000) -> None:
self(turn_limit = turn_limit
self.token_limit = token_limit
def check(self, turn_count: int, accumulated_tokens: int) -> None:
if turn_count > self.turn_limit:
self._trip("turn_ceiling", turn_count, accumulated_tokens)
if accumulated_tokens > self.token_limit:
self._trip("token_ceiling", turn_count, accumulated_tokens)
def _trip(self, reason: str, turn_count: int, accumulated_tokens: int) -> None:
print(
"\n=== 断路器检查点 ===\n"
f"异常原因 : {reason}\n"
f"当前调用次数 : {turn_count} / 最大限制 {self.turn_limit}\n"
f>总消耗令牌数 : {accumulated_tokens} / 最大限制 {self.token_limit}\n"
"处理措施 : 停止循环,由人类审核员介入\n"
"=================================="
)
raise CircuitBreakerError(
reason=reason,
turn_count=turn_count,
accumulated_tokens=accumulated_tokens,
)
CircuitBreakerError是一种异常,而不是返回码。这种设计是有意为之的:返回码可以被忽略,但未捕获的异常却不能被忽视。因此,隐藏异常的状态是不可能的。_trip()函数会在异常被抛出之前将那些便于人类阅读的检查点信息输出到标准输出中,所以即使调用者试图掩盖这个异常,操作员仍然能够了解到系统的状态。
一个非常重要的规则是:在每次调用大型语言模型之前,必须先调用.check()函数,而不是之后。如果在执行完操作后才进行检查,那就意味着在发现超出限制之前,你已经消耗了所有的令牌。
# 错误的做法——在执行后进行检查
result = client.messages.create(...)
breaker.check(turn_count, accumulated_tokens) # 这已经太晚了
# 正确的做法——在执行前进行检查
breaker.check(turn_count, accumulated_tokens) # 这会在任何操作发生之前就触发检查
result = client.messages.create(...)
默认的设置(5次循环,15,000个令牌)适用于简单的教程演示。但在实际生产环境中,你的资源预算可能是不同的,因此需要在创建对象时根据实际情况进行调整:
# 生产环境示例——减少令牌数量,增加循环次数
breaker = CircuitBreaker(turn_limit=10, token_limit=50000)
第三阶段:记录一切
断路器的作用是保护你的“资金账户”,而账本的作用则是帮助你清晰地了解究竟发生了什么。
大多数团队会进行日志记录以便于调试——他们希望在出现问题后能够查明原因。但账本的目的有所不同,它的作用在于实现对整个流程的监管。账本中的每一条记录都能证明该循环是否在规定的范围内进行,以及具体是在什么时候超出了范围。
# ledger.py
from ledger import Ledger
ledger = Ledger(db_path="ledger.db")
ledger.write(
session_id=spec.session_id,
turn_count=1,
state_origin="llm",
input_str=task,
token_delta=523,
execution_time_ms=1240,
pass_fail=True,
)
每次循环都会在账本中记录一条数据。这个账本是只支持追加操作,不允许进行更新或删除操作。这种不可变性的设计正是其核心所在:如果一个账本可以被随意修改,那么它就不再是一个真正的账本了,而更像是一本笔记。
账本的数据库结构如下:
CREATE TABLE IF NOT EXISTS ledger (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
turn_count INTEGER NOT NULL,
state_origin TEXT NOT NULL,
input_hash TEXT NOT NULL,
token_delta INTEGER NOT NULL,
execution_time_ms INTEGER NOT NULL,
pass_fail INTEGER NOT NULL, -- 1表示通过,0表示失败
breach_reason TEXT, -- 除非断路器被触发,否则这个字段为NULL
created_at TEXT NOT NULL -- 使用ISO 8601格式,时间戳以UTC为准
);
CREATE INDEX IF NOT EXISTS idx_ledger_session ON ledger(session_id);
通过创建索引,即使账本中的数据量逐渐增加,get_session(session_id)这种主要的查询操作也能保持恒定的查询速度。
按会话检索:
rows = ledger.get_session(spec.session_id)
for row in rows:
print(f"当前轮次:{row.turn_count}:{'通过' if row.pass_fail else '失败'} "
f"| {row.token_delta} 个代币 | {rowexecution_time_ms} 毫秒")
阶段 4:遵循自身界限的循环
代理循环将这三个基本组件连接在一起。它是唯一会调用大型语言模型的组件,其他所有操作都是在本地进行的。
# agent_loop.py
from agent_loop import AgentLoop
loop = AgentLoop(spec, breaker, ledger, client)
result = loop.run(task)
# LoopResult(success, turns, total_tokens, session_id, breach_reason)
一个轮次的执行流程如下:
在每次调用大型语言模型之前,都会进行预先检查,绝不例外。
def run(self, task: str) -> LoopResult:
session_id = self.spec.session_id
messages: list[dict] = [{"role": "user", "content": task}]
turn = 0
total_tokens = 0
try:
while True:
turn += 1
self.circuit_breaker.check(turn, total_tokens)
started = time.perf_counter()
response = self.client.messages.create(
model=self.model,
max_tokens=self.max_tokens,
system=self._system_prompt(),
messages=messages,
)
elapsed_ms = int((time.perf_counter() - started) * 1000)
turn_tokens = (
getattr(response.usage, "input_tokens", 0)
+ getattr(response_usage, "output_tokens", 0)
)
total_tokens += turn_tokens
text = self._text_from(response)
messages.append({"role": "assistant", "content": text})
self.ledger.write(
session_id=session_id,
turn_count=turn,
state_origin="llm",
input_str=task,
token_delta=turn_tokens,
execution_time_ms=elapsed_ms,
pass_fail=True,
)
if getattr(response, "stop_reason", "end_turn") == "end_turn":
return LoopResult(
success=True,
turns=turn,
total_tokens=total_tokens,
session_id=session_id,
)
messages.append({"role": "user", "content": "继续")
except CircuitBreakerError as err:
self.ledger.write(
session_id-session_id,
turn_count=turn,
state_origin="circuit_breaker",
input_str=task,
token_delta=0,
execution_time_ms=0,
pass_fail=False,
breach_reason=err.reason,
)
return LoopResult(
success=False,
turns=turn,
total_tokens=total_tokens,
session_id-session_id,
breach_reason=err.reason,
)
def _system_prompt(self) -> str:
return (
"你是一个正在执行特定任务的代理。\n\n"
f“这个任务的作用是:{self.spec.what_it_does}\n”
f“这个任务不能完成的事情是:{self.spec.what_it_does_not}\n”
f“完成任务后的结果应该是:{self/spec.done_looks_like}\n”
)
@staticmethod
def _text_from(response) -> str:
content = getattr(response, "content", None)
if not content:
return ""
block = content[0]
return getattr(block, "text", "") or ""
有几点值得特别强调:
LoopResult.session_id是从spec.session_id继承而来的。账本数据与规范文件之间无需通过复杂的连接操作就能建立对应关系,一个会话ID就代表整个可追踪的运行过程,从开始到结束。
第5阶段:审核界面
断路器可以保护你的银行账户安全,账本记录则能记录发生了什么事情。但这两者都无法告诉你实际发生的情况是否与你最初做出的承诺相符。
正是这种信息缺口导致了不良循环被批准通过——输出结果看起来很完美,仪表盘上也显示一切正常,但实际上并没有履行最初的承诺。审核人员看到这些结果后,认为它们是可以接受的,于是就签署了审批意见。但根本没有人去核实当初的承诺是否真正得到了履行。
审核界面就是为了填补这一信息缺口而存在的。它从SQLite数据库中读取相关数据,组装出包含五个元素的审核框架,并在任何下游系统接收输出结果之前,强制进行对比检查。
from review_surface import ReviewSurface
rs = ReviewSurface(spec_db_path="spec.db", ledger_db_path="ledger.db")
print(rs.render(session_id))
以下是这个包含五个元素的审核框架的具体内容,按顺序排列如下:
<当审核人确认满意时,他们会出具相应的证明:>
attestation = rs.attest(
session_id=result.session_id,
reviewer="daniel",
notes="输出结果符合规范。已批准。"
)
print(attestation.frame_hash)
.attest() 方法会将数据写入 ledger.db 文件中的 attestations 表中。frame_hash 是对规范框架数据的 SHA-256 散列值——对于审核同一会话内容的不同审核人员来说,这个散列值是相同的。它相当于一种审计凭证,能够证明审核人员看到的确实是原始数据,而不是任何摘要或改写内容。
批准操作确认了整个流程已经执行完毕;而认证过程则证明了审核人员确实将输出结果与初始承诺进行了对比。当这个流程涉及到受监管的内容时,所产生的文件就会属于不同的法律类别。
@dataclass(frozen=True)
class ReviewFrame:
session_id: str
original_promise: SpecResult
acceptance_criteria: str
diff: DiffResult
evidence: tuple # 元组形式,元素为LedgerRow类型
unresolved_assumptions: tuple # 元组形式,元素为str类型
created_at: str
ReviewFrame> 类被设为不可修改的状态,原因与 SpecResult> 类相同——因为这些数据属于证据,而不是草稿。evidence 和 unresolved_assumptions 被定义为元组,是因为列表无法被哈希处理,而不可修改的数据结构需要使用可哈希的字段。
关于审核流程的完整端到端实现代码可以在仓库中的 examples/review_example.py 文件中找到。在任何会话完成之后运行这个脚本,它就会生成包含五个元素的框架数据,提示用户进行认证操作,如果你批准的话,还会生成相应的审计凭证。
整个流程需要你的确认才能继续执行;在得到你的批准之前,下游系统是无法收到任何处理结果的。
阶段 6:一个实际案例——SEO审核代理
只有面对真实的问题时,这种设计模式才有意义。我的 seo-agent 项目就是基于这种架构设计的。
SEO审核工作本身具有固定的节奏:首先爬取网站数据,找出存在的问题并进行修复,然后等待搜索引擎重新索引这些内容。如果持续运行这个审核代理程序,也不会改变这一节奏;它只会浪费一些资源,在那些真正重要的环节之间消耗多余的“代币”。将定时任务与这个流程结合使用,才是合理的架构设计。
# examples/seo_audit_example.py
import requests
from bs4 import BeautifulSoup
import anthropic
from spec_writer import SpecWriter
from circuit_breaker import CircuitBreaker
from ledger import Ledger
from agent_loop import AgentLoop
def crawl_url(url: str) -> str:
response = requests.get(url, timeout=10)
soup = BeautifulSoup(response.text, "html.parser")
title = soup.find("title")
meta_desc = soup.find("meta", attrs={"name": "description"})
h1_tags = soup.find_all("h1")
return (
f"URL: {url}\n"
f"标题: {title.text if title else '未找到'}\n"
f>元描述: {\n}{meta_desc['content'] if meta_desc else '未找到'}\n"
f>"h1标签数量: {len(h1_tags)}\n"
f>"h1标签内容: {[h.text[:50] for h in h1_tags]}"
)
def run_seo_audit(url: str) -> None:
# 第一步:在流程开始之前定义完成标准
spec = SpecWriter(db_path="spec.db").run()
# 第二步:初始化断路器和账本
breaker = CircuitBreaker(turn_limit=5, token_limit=15000)
ledger = Ledger(db_path="ledger.db")
client = anthropic.Anthropic()
# 第三步:爬取指定网站的数据
site_data = crawl_url(url)
# 第四步:运行审核流程
# AgentLoop会内部捕获CircuitBreakerError异常并返回相应的结果
# 因此不需要使用try/except语句来处理CircuitBreakerError异常
loop = AgentLoop(spec, breaker, ledger, client)
result = loop.run(
f"请审核此页面的SEO问题:\n\n{site_data}"
)
# 第五步:输出审核结果
print(f"\n结果: {'SUCCESS' if result.success else '失败'}")
if not result.success:
print(f>失败原因: {result.breach_reason}")
print(f>尝试次数: {result.turns} | 代币消耗量: {result.total_tokens}")
print("\n审核记录:")
for row in ledger.get_session(result.session_id):
status = "通过" if row.pass_fail else "失败"
print(f" 尝试次数 {row_turn_count}: {status} | "
f"消耗代币数 {row.token_delta} | 执行耗时:{rowexecution_time_ms}毫秒")
if __name__ == "__main__":
import sys
run_seo_audit(sys.argv[1] if len(sys.argv) > 1 else "https://example.com")
运行它:
python examples/seo_audit_example.py https://yourdomain.com
规范编写工具会引导你完成操作。循环会持续运行,如果超过了限制,断路器就会启动;同时,账本会记录下每一次操作的结果。最终输出结果会显示在你面前,你可以据此决定需要修改哪些地方。
这个循环是为你服务的,而不是在无意义的空虚中运行。
可插拔的LLM客户端
这个循环可以与任何符合LLMClient协议的客户端配合使用(默认使用的是Anthropic)。你也可以通过编写大约20行的适配代码,使用自己定制的客户端。
# agent_loop.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class MessagesEndpoint(Protocol):
def create(self, *, model: str, max_tokens: int,
system: str, messages: list) -> object: ...
@runtime_checkable
class LLMClient(Protocol):
messages: MessagesEndpoint
messages被定义为一个实例属性(而不是嵌套类),因为真实的Anthropic SDK就是这么设计的——通过anthropic.Anthropic().messages.create(...)来使用这个接口。如果将其定义为嵌套类,那么真实的客户端就无法满足LLMClient协议的要求了。@runtime_checkable装饰器可以帮助我们通过isinstance(client, LLMClient)来检查代码是否符合规范,而仓库中的测试用例也是利用这个判断方法来检测FakeClient这个模拟客户端的。
下面是一个OpenAI适配器的示例(这个示例仅用于说明目的,实际生产环境中使用的适配器还需要处理数据流、工具使用方式以及错误信息等问题):
# openai_adapter.py — 仅为说明用途的伪代码,并非可用于生产环境。
from openai import OpenAI as _OpenAI
class _MessagesAdapter:
def __init__(self, client):
self._client = client
def create(self, *, model, max_tokens, system, messages):
completion = self._client.chat.completions.create(
model=model,
max_tokens=max_tokens,
messages=[{"role": "system", "content": system}] + messages,
)
# 将OpenAI返回的结果转换成Anthropic可以处理的格式
# AgentLoop会接收如下格式的响应:response.usage.{input,output}_tokens,
# response.content[0].text, response.stop_reason.
return _adapt_responsecompletion)
class OpenAIAdapter:
def __init__(self, api_key: str):
self._client = _OpenAI(api_key=api_key)
self.messages = _MessagesAdapter(self._client) # 这是一个实例属性,而不是嵌套类
适配器模式确实值得专门进行讲解。因为不同的提供者API其数据结构是不一样的。Anthropic将system信息放在最顶层,而OpenAI则将其包含在messages数组中。通过编写大约20行的适配代码,我们就可以让这个循环机制与不同的提供者API兼容,而无需重新修改任何代码。需要注意的是,self.messages是在__init__方法中被初始化的,因此它是每个适配器实例的真实属性,其数据结构与真实的SDK是一致的。
运行测试
python -m pytest tests/
要查看代码覆盖率,请执行以下命令:
python -m coverage run --source=circuit_breaker,ledger,spec_writer,agent_loop,review_surface -m pytest tests/
python -m coverage report -m
共有80个测试用例,所有5个核心模块的代码覆盖率均为100%。在tests/test_agent_loop.py文件中,使用了FakeClient作为测试对象来执行相关测试。这种设计通过“鸭子类型”机制满足了LLMClient协议的要求:将messages属性设置为self,因此client.messages.create(...)方法会调用同一个对象,并为每种测试场景生成预设的响应结果。只需克隆该代码仓库并运行pytest命令,就能看到所有80个测试用例都能成功通过,而且无需连接网络或使用API密钥。
circuit_breaker.py文件的代码覆盖率也为100%,没有任何未经过测试的路径。该模块是负责确保系统财务安全的关键组件,其中的所有功能路径都经过了测试。
你所构建的内容
通过本教程,你构建了5个独立可用的小型模块。
| 模块名称 | 作用 | 代码行数 |
|---|---|---|
spec_writer.py |
在循环开始之前强制设置三个固定答案 | 104行 |
circuit_breaker.py |
对轮次执行次数和代币数量进行严格限制 | 41行 |
ledger.py |
仅支持追加操作的SQLite审计日志系统 | 113行 |
agent_loop.py |
一个能够同时满足各种要求的循环机制 | 128行 |
review_surface.py |
用于组装包含5个元素的数据结构并记录人工验证信息 | 114行 |
这种设计模式的核心在于:上游模块负责设定规则边界,下游模块则负责执行这些规则;两者都不依赖模型来自我监管。
一个没有退出条件的循环并不具备自主性,它实际上只是个等待被执行的操作而已。
在开始开发之前,就必须明确“完成目标应该是什么样子”。这才是真正重要的工作,而且这一点从来都没有改变。
后续步骤
该代码仓库的地址是github.com/dannwaneri/production-safe-agent-loop。
如果你想进一步深入研究,还有三个自然的扩展方向:
1. 向分布式系统发展
对于那些仅用于独立序列执行的系统来说,SQLite数据库是完全可以满足需求的。但一旦有多个代理进程共享同一数据状态,就需要使用可序列化的隔离机制——因为对扁平化JSON数据的并发写入操作可能会导致数据被破坏。README文件中详细说明了在哪些情况下系统需要升级到分布式架构。
2. 加密签名技术
对于那些在运行过程中没有审计人员在场的环境来说,仅使用SQLite数据库是不够的。数据库管理员可以执行UPDATE操作来修改数据,但这种做法存在安全隐患。而使用Ed25519加密算法为每条记录添加签名,则可以确保日志内容在执行后被妥善保护。不过,这个话题属于另一个教程的内容了。
配置 Cron 作业
实际上,SEO审计代理的真正工作方式并不是全天候自动运行。它是由 cron 作业按照预定时间表来执行的,当发现异常情况时才会停止运行。0 3 * * 2 python examples/seo_audit_example.py https://yourdomain.com——这就是整个配置流程。这个循环会持续运行,而不会陷入无休止的重复状态。
如果您需要为自己的系统架构配置类似的机制(例如断路器、审计日志功能,或是确保代理程序在生产环境中能够安全运行的机制),我愿意提供自由职业服务。dannwaneri.com/ai-agents/