循环编程：用于自我提升的AI智能体的四个嵌套循环
用于自我提升的AI智能体的循环设计：四个嵌套循环（ReAct、重试、队列轮询器、爬山算法），再加上一个评估器、预算计数器以及篡改检测机制。
来源: Loop Engineering: Building Self-Improving AI Agents with Four Nested Loops
大约两年来，与模型协作一直意味着轮流操作。工程师编写提示词，查看模型返回的结果，然后手动输入下一部分上下文。这种工作模式存在隐性成本，即一种协调成本，因为整个流程的速度受限于人类审核的速度。要突破这一限制，就需要转变思路：不再直接操作工具，而是开始设计能够承载该工具的系统。
有三个理念为这种转变提供了框架。
- 影响力上升了一个层级。角色的职能从编写精确指令转变为设计重复执行的循环，最终类似于负责整体规划的角色。人类负责设定目标与停止条件，而智能体则掌控两者之间的每一步流程。
- 自主循环具有可辨识的结构。它们会寻找任务，将工作分配给子代理，通过独立测试检查结果，并记录自身状态，从而让人类无需再参与分类、修正和验证的循环。
- 目标设计处于核心位置。在此层面，循环的内部运作细节被弱化。打破自1969年以来矩阵乘法记录的AlphaEvolve，以及能将高层目标转化为具体要求和规范的亚马逊Kiro，都体现了这一模式：只要设计出正确的评分函数，机器就能找到答案。
要为智能体设计循环结构，首先需要对其中的智能体有清晰的心理模型。因此，合理的构建顺序是：从一个简单的工具调用工作器开始，再添加一个用于评估其输出结果的模型（这样系统就不会再自我评分），将两者通过循环连接起来，并将评估结果写入磁盘——因为模型会遗忘，而存储库不会。之后需要设计防止智能体操纵评估指标的机制，也就是在METR测试OpenAI的o3时所发现的那种作弊行为；接着要开发一个能在Token数量多或少的环境中跟踪成本的指标；最后则是将原本只有70行的脚本逐步发展成能够在实际环境中稳定运行的版本。
以下便是基于保险核保流程作为测试平台，对整个理念的实际实现示例。
为何这属于架构问题
从能够回答问题的模型转变为能够可靠执行任务的智能体，并非简单的提示词编写问题，而是一个设计难题。大多数教程展示的只是单个智能体调用单个工具的场景，而实际系统需要同时具备容错能力、自我修正功能、支出限制、篡改检测以及持续改进的能力。
这里采用的方法被称为循环工程：通过嵌套多个具有不同功能的控制循环，使整个系统的稳定性优于任何一个单独的循环。
保险核保是一个合适的应用领域——决策选项有限且固定（接受、修改、拒绝、转介），规则明确，历史数据可提供正确与错误的参考答案。这些模式同样适用于代码审查、文档分析、支持服务分级处理以及医疗记录处理。
四个相互嵌套的循环
普通软件通过多层抽象来掌控复杂性，而循环工程则通过多层反馈机制来应对智能体的不可靠性：
┌──────────────────────────────────────────────────────┐
│ Loop 4 — Hill Climbing (self-improvement over time) │
│ ┌────────────────────────────────────────────────┐ │
│ │ Loop 3 — Event Queue Poller (orchestration) │ │
│ │ ┌──────────────────────────────────────────┐ │ │
│ │ │ Loop 2 — Verification & Retry │ │ │
│ │ │ ┌────────────────────────────────────┐ │ │ │
│ │ │ │ Loop 1 — ReAct Agent │ │ │ │
│ │ │ │ (think → act → observe → think) │ │ │ │
│ │ │ └────────────────────────────────────┘ │ │ │
│ │ └──────────────────────────────────────────┘ │ │
│ └────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────┘
每个循环都有其特定的职责、需要防范的故障类型以及会响应的反馈信号。关键在于防止不同循环之间的问题相互干扰。
领域模型
在讨论架构之前，先来看数据。保险申请表是一份结构简单的记录：
{
"id": "APP-001",
"applicant_age": 34,
"state": "FL",
"property_type": "residential",
"coverage_amount": 450000,
"claim_history_5yr": 1,
"credit_score_tier": "good",
"flood_zone": true,
"has_flood_rider": false,
"business_use": false
}
这20个示例应用每个都有一个真实标签：
"APP-001": {
"split": "train",
"expected_decision": "modify",
"expected_flags": ["flood_rider_required", "windstorm_exclusion"]
}
决策共有四种类型：
- 接受：风险在可接受范围内，按现状批准
- 修改：附加条件或备注后批准
- 拒绝：风险超出可接受阈值
- 转交：需由高级核保员处理
一个小型知识库中存储着五条隐含的核保规则，涉及洪水区域、理赔历史、信用评分、保障限额以及各州的特定限制。系统提示中不会告知这些规则，代理必须通过工具调用来查找并应用它们。训练集包含16份申请，另有4份用于测试。
循环1：ReAct代理
防御对象：那种在没有证据的情况下凭空做出决策的模型。反馈信号：工具调用图，即智能体在做出决策前是否查阅了所有相关信息。
ReAct是“推理加行动”的缩写，属于基础模式。智能体会先思考自己的需求，调用相应工具，读取结果，再次思考，如此循环直至做出最终决定：
System Prompt
│
▼
[Think] → What information do I need?
│
▼
[Act] → Call tool (lookup_application, retrieve_rules, ...)
│
▼
[Observe] → Tool returns result
│
▼
[Think] → What does this mean? What next?
│
... (repeat until decision is made)
│
▼
[Act] → draft_decision (terminal tool call)
在LangGraph中，这会变成一个包含两个节点和一条条件边的状态图。该状态包含应用ID、消息列表、决策字段、学习内容、当前支出以及步数计数器：
class UnderwritingState(TypedDict):
application_id: str
messages: Annotated[list[AnyMessage], add_messages] # reducer accumulates
decision: str
rationale: str
flags: list[str]
recommended_premium_adj: float
lessons: list[str]
spend: float
_steps: int
有一个细节值得注意。add_messages注解使得LangGraph在每一步中都将新消息追加到消息列表中，而非替换原有内容，这样同一个应用程序在每次工具调用时都能保留完整的对话历史。
构建器会获取系统提示词和共享的成本计量器，然后将各类工具与模型绑定在一起：
def build_loop1(system_prompt: str, meter: CostMeter):
model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools(TOOLS)
其中包含代理节点、工具节点、路由功能以及图结构组件：
def agent_node(state: UnderwritingState) -> dict:
prompt = system_prompt
if state.get("lessons"):
prompt += "\n\nUnderwriting lessons:\n" + "\n".join(
f"- {l}" for l in state["lessons"]
)
msgs = [SystemMessage(content=prompt)] + list(state["messages"])
response = model.invoke(msgs)
meter.record_from_message(response) # track spend per call
return {
"messages": [response],
"_steps": state.get("_steps", 0) + 1,
"spend": meter.spent,
}
def tools_node(state: UnderwritingState) -> dict:
last = state["messages"][-1]
tool_results, updates = _dispatch_tools(last.tool_calls)
return {"messages": tool_results, **updates} # updates captures draft_decision output
def should_continue(state) -> Literal["tools", "__end__"]:
if state.get("_steps", 0) >= MAX_STEPS: # hard cap at 8 steps
return END
if getattr(state["messages"][-1], "tool_calls", None):
return "tools"
return END
g = StateGraph(UnderwritingState)
g.add_node("agent", agent_node)
g.add_node("tools", tools_node)
g.set_entry_point("agent")
g.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
g.add_edge("tools", "agent")
return g.compile()
四种工具
该智能体恰好拥有四种工具，足以满足其所有需求：
@tool
def lookup_application(app_id: str) -> dict:
"""Return the full application record for the given application ID."""
# reads from applications.json — deterministic, no LLM call
...
@tool
def retrieve_rules(risk_factor: str) -> list[str]:
"""Return underwriting rules matching the given risk factor keyword.
Use terms like: flood, claims, credit, coverage, state, business."""
# keyword match against rules_kb.json - forces explicit rule lookup
...
@tool
def fetch_risk_signal(signal_type: str) -> dict:
# simulated external data pull (credit bureaus, flood maps, etc.)
...
@tool
def draft_decision(
decision: Literal["accept", "modify", "decline", "refer"],
rationale: str,
flags: list[str],
recommended_premium_adj: float,
) -> dict:
"""Finalize the underwriting decision with structured output."""
# terminal action - structured schema forces the agent to commit explicitly
...
lookup_application仅用于读取数据。retrieve_rules会在固定的规则集中进行关键词搜索，从而强制进行显式查找。fetch_risk_signal则模拟从外部数据源（如信用机构或洪水地图）获取数据。draft_decision是最终操作，也是最重要的环节——它会让智能体基于理由和特定标记做出明确的决策，因此结构化的结果直接来自工具调用本身，无需再解析自由文本。
为何步骤上限很重要
MAX_STEPS = 8的作用是防止工具链无限循环。一个混乱的模型可能会反复二十次调用“洪水”相关的规则查找功能，试图找到根本不存在的答案，从而浪费令牌和时间。当步骤数达到上限时，系统状态会变为ran_out_of_steps，循环2可以识别这一状态并重新尝试。
循环2：验证与重试
防御目标：避免因推理错误而出现单个错误答案。反馈机制：由确定性判断函数is_correct()提供反馈，无需依赖模型。
循环1可能会失败。智能体可能误读标记、跳过规则或达到步骤上限。循环2能捕捉这些故障，并通过有针对性的反馈让智能体再次尝试：
def run_loop2(
app_id: str,
system_prompt: str,
meter: CostMeter,
max_retries: int = 3,
) -> tuple[dict, bool]:
state = run_loop1(app_id, system_prompt, meter)
for attempt in range(max_retries):
decision = state.get("decision", "ran_out_of_steps")
if decision == "refer":
_write_human_queue(app_id, state) # human-in-the-loop path
return state, True
if is_correct(decision, app_id): # deterministic check
return state, True
if attempt >= max_retries - 1:
return state, False # exhausted retries
# construct targeted feedback for next attempt
expected = load_historical_decisions()[app_id]["expected_decision"]
feedback = HumanMessage(content=(
f"Your decision was '{decision}' but this application requires '{expected}'. "
f"Review the risk factors carefully and call draft_decision again."
))
extra_messages = list(state.get("messages", [])) + [feedback]
state = run_loop1(app_id, system_prompt, meter, extra_messages=extra_messages)
return state, False
有几点设计选择值得详细说明。
裁判是确定性的。 is_correct()会读取历史决策文件并比对字符串，从不调用模型。若使用第二个模型作为裁判，只会增加另一处不可靠因素，而此处的事实标准本身就是最准确的依据。
转交处理属于已结束的结果。当智能体发出refer指令时，循环2会将该案例写入human_queue.json并报告处理成功。转交给人工处理被视为有效的结束状态，无需重新尝试。
反馈会明确指出目标。消息不会只说“再试一次”，而是告知智能体哪类决策是正确的，并要求其重新审视风险因素。这样能引导下一次运行更接近正确答案，而盲目重新采样则可能偏离正确方向。
历史记录会被保留。额外的消息包含了之前的全部对话内容以及反馈信息，因此当代理再次运行时，它能看到之前尝试过的一切，从而减少重复走错路径的可能性。
循环3：事件队列轮询器
防止的问题：任务生成者与任务处理者之间的紧密耦合。
反馈信号：pending_applications.json文件的内容，该文件作为队列使用。
这个循环充当协调者角色。它从待处理队列中取出任务，驱动某个应用依次经过循环1和循环2，更新运行状态，并偶尔触发循环4：
def run_event_loop(
system_prompt: str,
meter: CostMeter,
pending_path: Path = DEFAULT_PENDING,
state_path: Path = DEFAULT_STATE,
once: bool = False,
) -> None:
run_state = load_run_state(state_path)
verify_judge_checksum(run_state.get("judge_checksum", "")) # tamper check
total = len(json.loads(pending_path.read_text(encoding="utf-8-sig"))
if pending_path.exists() else [])
processed = 0
while True:
pending = (json.loads(pending_path.read_text(encoding="utf-8-sig"))
if pending_path.exists() else [])
if not pending:
if once:
print("[loop3] queue empty - done")
break
print("[loop3] queue empty - sleeping...")
time.sleep(POLL_INTERVAL)
continue
app_id = pending[0]
processed += 1
print(f"[loop3] {processed}/{total} {app_id} ...", flush=True)
try:
result_state, passed = run_loop2(app_id, prompt_with_lessons, meter)
except BudgetExhaustedError:
raise # budget exhaustion is fatal
except Exception as exc:
print(f"[loop3] {app_id} ERROR: {exc!r} - skipping")
# remove from queue and continue - one bad app cannot block the rest
remaining = [x for x in pending if x != app_id]
pending_path.write_text(json.dumps(remaining), encoding="utf-8")
continue
# ... update state, distill lesson, trigger hill-climbing
为何使用文件作为队列
有意选择JSON文件而非Redis流或SQS队列，是因为在本地开发时，这样的格式便于查看（打开文件即可了解待处理任务），便于恢复（崩溃后重启，队列内容依然完整），且具有解耦性（填充队列的程序无需了解消费者信息）。在生产环境中则会使用专业的消息中间件，而接口契约——即ID列表——则保持不变。
篡改检测
在处理任何内容之前，Loop 3会先验证真实数据：
verify_judge_checksum(run_state.get("judge_checksum", ""))
支票的运作方式如下：
def compute_judge_checksum() -> str:
content = _HIST.read_bytes() # historical_decisions.json
return hashlib.sha256(content).hexdigest()
def verify_judge_checksum(expected: str) -> None:
actual = compute_judge_checksum()
if actual != expected:
raise RuntimeError(
f"Judge checksum mismatch - historical_decisions.json was tampered with."
)
在初始化时，历史决策文件的SHA-256哈希值会被存储在运行状态中，每次事件循环启动时都会重新校验。如果有人篡改真实数据以让智能体表现更好，系统就会拒绝运行。这堵上了自我改进系统中的一个隐蔽漏洞：提高得分的最廉价方法就是更改评分标准。
跳过处理与崩溃处理
第3轮中的try/except机制刻意划定了界限。预算耗尽时会再次触发该机制，从而停止所有操作。其他任何异常都会被记录下来，出问题的应用程序会被从队列中移除，然后处理下一个应用程序。因此，单个格式错误的应用程序无法冻结整个流程，而按设计，资金耗尽则是致命的，因为系统在没有任何可用资金的情况下不应继续运行。
第4轮：爬山算法
防御对象：从不从经验中学习的固定提示词。反馈信号：训练准确率以及审计结果。
每做出第八个决策时（HILL_CLIMB_EVERY = 8），循环3会将控制权交给循环4，后者通过反思、变异、评估和筛选的流程自动优化系统提示词：
seed prompt
│
▼
evaluate on training set → score, failures
│
▼
┌────────────────────────────────┐
│ for each round: │
│ │
│ reflect(current, failures) │ ← LLM analyzes failure patterns
│ │ │
│ ▼ │
│ candidate_prompt │
│ │ │
│ ▼ │
│ evaluate(candidate, train) │ ← full eval run
│ │ │
│ ▼ │
│ audit_prompt(candidate) │ ← anti-cheating check
│ │ │
│ ▼ │
│ if score > best AND clean: │
│ best = candidate │ ← keep
│ else: │
│ discard │ ← revert
└────────────────────────────────┘
│
▼
write best_prompt.txt
反思
REFLECT_TEMPLATE = """You are improving the system prompt of an insurance underwriting agent.
CURRENT PROMPT:
{prompt}
The agent got these decisions WRONG on training applications:
{failures}
Look for PATTERNS in the failures. Infer GENERAL underwriting rules that would fix them.
Do NOT memorize specific application IDs or applicant details.
Write an improved system prompt. Reply with ONLY the new prompt text."""
def reflect(current_prompt: str, failures: list[dict]) -> str:
client = anthropic.Anthropic()
failure_text = "\n".join(
f"- APP {f['app_id']}: agent said '{f['got']}', expected '{f['expected']}'. "
f"Application data: {f['application']}"
for f in failures
)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1500,
messages=[{"role": "user", "content": REFLECT_TEMPLATE.format(
prompt=current_prompt,
failures=failure_text,
)}],
)
return response.content[0].text.strip()
不得记忆特定应用编号或申请人详细信息的指令至关重要。如果没有这一限制，模型可能会生成诸如“对于APP-003，始终拒绝”的提示语，这类提示语在训练数据上能获得完美得分，但在未见过的案例中则会彻底失效。新的提示语必须从失败案例中的规律中归纳出通用规则，而非依赖个别案例。
评估
def evaluate_prompt(system_prompt: str, split: str = "train", meter=None):
tasks = load_tasks(split)
failures = []
for task in tasks:
state = run_loop1(task["id"], system_prompt, meter)
decision = state.get("decision", "ran_out_of_steps")
if not is_correct(decision, task["id"]):
failures.append({
"app_id": task["id"],
"got": decision,
"expected": task["expected_decision"],
"application": task["application"],
})
score = (len(tasks) - len(failures)) / len(tasks)
return score, failures
每个候选提示词都会在全部16个训练实例上经过第一轮循环测试。这成本很高，每轮需要运行16次模型，五轮下来至少需要80次，但这样的测试结果具有真实性：一个提示词必须能在整个训练数据集中有效，而不仅仅在某个样本上有效。
筛选机制
def improve(seed_prompt: str, lessons: list[str], meter: CostMeter, rounds: int = 5) -> str:
best = with_lessons(seed_prompt, lessons)
best_score, failures = evaluate_prompt(best, split="train", meter=meter)
for r in range(1, rounds + 1):
candidate = reflect(best, failures)
score, cand_failures = evaluate_prompt(candidate, split="train", meter=meter)
verdict = audit_prompt(candidate) # anti-cheat check
if score > best_score and verdict == "VERDICT: CLEAN":
best, best_score, failures = candidate, score, cand_failures
BEST_PROMPT_PATH.write_text(best, encoding="utf-8")
print(f"round {r}: {score:.0%} KEPT ({len(cand_failures)} failures)")
else:
reason = "cheating" if verdict != "VERDICT: CLEAN" else f"{score:.0%} <= {best_score:.0%}"
print(f"round {r}: {score:.0%} REVERTED ({reason})")
return best
在接纳某个候选答案之前，必须同时满足两个条件。首先，score > best_score，即其得分必须严格高于当前最佳值；出现平局的情况将被拒绝，以此避免提示词偏移。其次，审核员必须返回VERDICT: CLEAN，以确认该候选答案中不含硬编码的答案。
提示词审核器
AUDIT_SYSTEM = (
"You are a security auditor for AI systems. "
"Check whether the prompt contains any hardcoded application IDs (e.g. APP-001) "
"or hardcoded decisions for specific applications. "
"These would constitute cheating — memorizing answers instead of learning rules. "
"Reply with exactly one of:\n"
"VERDICT: CLEAN\n"
"VERDICT: CHEATING\n"
"Followed by a one-sentence explanation."
)
def audit_prompt(evolved_prompt: str) -> str:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=200,
system=AUDIT_SYSTEM,
messages=[{"role": "user", "content": f"PROMPT TO AUDIT:\n{evolved_prompt}"}],
)
return response.content[0].text.strip().split("\n")[0] # first line only
这是一种以大语言模型作为评判者的设计，但仅限于一个二元结构化问题：提示语中是否包含特定编号或已记忆的答案？这类分类任务中，模型评判者是可靠的。审核者从不评估质量、风格或正确性，仅查找某种特定模式。
经验记忆
除了爬山算法外，该系统还会维护一个在多次运行间持续存在的lessons列表。当应用程序在多次重试后仍失败时，它会调用distill_lesson()函数：
DISTILL_PROMPT = """An insurance underwriting agent made a wrong decision.
Application details: {application}
Agent decided: {got}
Correct decision: {expected}
Write ONE short, general underwriting rule that would prevent this mistake.
Do NOT reference this specific application ID or any applicant names/details.
Reply with only the rule, as a single sentence."""
def distill_lesson(application: dict, got: str, expected: str) -> str:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=150,
messages=[{"role": "user", "content": DISTILL_PROMPT.format(...)}],
)
return response.content[0].text.strip()
每节课都会通过 with_lessons() 被添加到后续的每一次 Loop 1 运行中：
def with_lessons(base_prompt: str, lessons: list[str]) -> str:
if not lessons:
return base_prompt
lessons_text = "\n".join(f"- {l}" for l in lessons)
return base_prompt + f"\n\nUnderwriting lessons learned:\n{lessons_text}"
这是一种无需梯度下降的少样本学习方法。系统在运行时通过收集以自然语言规则形式呈现的压缩知识来从错误中学习。一个类似CL-BENCH协议的函数用于检测这些经验是否真的能起到帮助作用：
def measure_gain(base_prompt: str, lessons: list[str], split: str = "test", meter=None) -> float:
stateless_score, _ = evaluate_prompt(base_prompt, split=split, meter=meter)
augmented = with_lessons(base_prompt, lessons)
stateful_score, _ = evaluate_prompt(augmented, split=split, meter=meter)
return stateful_score - stateless_score # positive = lessons help
利用成本计量器进行预算控制
每次模型调用都会经过一个共享的 CostMeter，它从 Loop 4 传递到 Loop 3，再传递到 Loop 2，最后传递到 Loop 1，因此所有支出都会集中在一处：
BUDGET_USD = 2.00
INPUT_PRICE_PER_TOKEN = 3.00 / 1_000_000
OUTPUT_PRICE_PER_TOKEN = 15.00 / 1_000_000
class CostMeter:
def record(self, input_tokens: int, output_tokens: int) -> None:
if self.spent >= self._budget: # check BEFORE adding
raise BudgetExhaustedError(
f"Budget ${self._budget:.2f} exhausted at ${self.spent:.4f}"
)
self.spent += input_tokens * INPUT_PRICE_PER_TOKEN \
+ output_tokens * OUTPUT_PRICE_PER_TOKEN
顺序至关重要：在添加成本之前先检查预算。这样一旦达到限额就会立即触发错误，而不会在已经超支之后才发现。BudgetExhaustedError会沿着循环层级向上传递，Loop 3会重新抛出该错误而非将其吞没，因此预算耗尽总会终止整个运行过程。
具有幂等设置的运行状态
run_state.json记录了所有发生过的内容：已处理的申请、通过和失败的列表、积累的经验、自上次调整后的决策以及校验和。它的初始化方式具有防御性：
def _init_run_state() -> dict:
path = _STATE_DIR / "run_state.json"
state = {}
if path.exists():
try:
content = path.read_text(encoding="utf-8-sig").strip() # handles Windows BOM
if content:
state = json.loads(content)
except (json.JSONDecodeError, UnicodeDecodeError):
pass # corrupt file → start fresh
if not state:
state = dict(_DEFAULT_RUN_STATE)
state["judge_checksum"] = compute_judge_checksum() # always refresh
path.write_text(json.dumps(state, indent=2), encoding="utf-8")
return state
这里体现了三种增强韧性的习惯。使用 utf-8-sig 进行读取时会自动移除 Windows 的字节顺序标记。遇到空文件或损坏的文件时，程序会恢复到默认状态而不会崩溃。此外，每次启动时都会重新计算校验和，绝不会依赖上一次运行时的结果。
三种运行方式
共有三种启动方式，每种方式仅使用堆栈的一部分功能。
单模式仅运行循环 1，这便于调试单个应用程序，并查看代理调用了哪些工具以及原因：
python -m src.main --mode single --app-id APP-001
事件循环模式会针对每个待处理的应用程序执行循环1、2和3，并且在每做出八次决策后可能会触发循环4。--once标志会清空队列并立即退出，而不会无限轮询：
python -m src.main --mode event-loop --once
改进模式会直接运行第4轮循环，在训练数据集中进行五轮爬坡优化。随后，它会使用保留的测试数据集评估找到的最佳提示词，并输出提升幅度：
python -m src.main --mode improve --rounds 5
一个应用，全程处理
这是单个应用在系统中流转的完整数据流：
pending_applications.json
│
│ [Loop 3 reads APP-009]
▼
run_loop2("APP-009", prompt, meter)
│
│ [Loop 2 calls Loop 1]
▼
run_loop1("APP-009", prompt, meter)
│
│ [LangGraph StateGraph]
▼
agent_node → ChatAnthropic.invoke([SystemMsg, HumanMsg])
│
│ response: tool_call(lookup_application, {"app_id": "APP-009"})
▼
tools_node → lookup_application.invoke({"app_id": "APP-009"})
│
│ returns: {id: APP-009, state: TX, coverage: 1200000, ...}
▼
agent_node → ChatAnthropic.invoke([..., ToolMessage])
│
│ response: tool_call(retrieve_rules, {"risk_factor": "coverage"})
▼
tools_node → retrieve_rules.invoke({"risk_factor": "coverage"})
│
│ returns: ["Coverage > $1M requires mandatory refer. Flag: high_coverage"]
▼
agent_node → ChatAnthropic.invoke([..., ToolMessage])
│
│ response: tool_call(draft_decision, {decision: "refer", ...})
▼
tools_node → draft_decision.invoke({...})
│
│ updates state: {decision: "refer", flags: ["high_coverage"], ...}
▼
should_continue → END (no more tool calls)
│
▼
Loop 2: decision == "refer" → write human_queue.json → return (state, True)
│
▼
Loop 3: passed=True → update run_state.json → remove APP-009 from queue
│
│ [decisions_since_last_improvement becomes 9 >= 8]
▼
Loop 4: improve(seed_prompt, lessons, meter, rounds=5)
│
│ reflect → evaluate → audit → gate → write best_prompt.txt
▼
Loop 3: continue with APP-010 using new best_prompt
构建过程中的经验教训
1. 将判定器与智能体分开。判定器（is_correct、evaluate_prompt）与智能体（run_loop1）不共享任何代码。判定器负责读取静态文件，而智能体则负责调用模型。让模型自行评估自己的输出，这样就能得到一个从不出错的附和者。
2>确保评估结果具有确定性。对于相同的输入，每次运行的正确性、训练准确率和测试准确率都是确定的。唯一的随机性来自模型的回复，因此只要有了提示词和种子，任何分数都可以被复现。
3. 使用最终答案工具以实现结构化输出。 draft_decision 工具比解析普通文本更有效。其架构规定了词汇表（accept | modify | decline | refer）、必填字段（flags、rationale、recommended_premium_adj）及其类型，因此智能体必须明确表达自己的决策。
4. 在自我改进系统中，反作弊并非可选功能。 一旦系统能够修改自身的指令，它就会寻找捷径，而最明显的捷径就是记住训练时的答案。audit_prompt() 可以自动检测这种情况。如果没有该功能，爬山法生成的提示语在训练数据上的准确率几乎完美，但在测试数据上的准确率则近乎随机。
5. 字节顺序标记与编码属于生产环境中的缺陷。如果在 Windows 系统上构建程序或接收来自 Windows 用户的文件，每次读取时都必须使用 encoding="utf-8-sig"。Windows 记事本生成的 BOM 会引发 json.JSONDecodeError: Unexpected UTF-8 BOM 错误，由于该文件在编辑器中看起来完全正常，因此很难诊断问题。
6. 应在支出前检查预算，而非事后。
# WRONG: allows one overspend before raising
self.spent += cost
if self.spent > self._budget:
raise BudgetExhaustedError(...)
# CORRECT: raises before the overspend registers
if self.spent >= self._budget:
raise BudgetExhaustedError(...)
self.spent += cost
当多个模型调用连续快速执行时，顺序很重要。先添加再后续检查，可能在出现错误之前，就已经有超过一次调用的资源消耗超出了预算。
7. 在长循环中刷新进度条。
print(f"[loop3] {processed}/{total} {app_id} ...", flush=True)
在 Windows 系统中，除非另有说明，否则标准输出会被缓冲。如果省略 flush=True，则“处理 APP-009”这条信息会在 API 调用返回后才出现，这样就失去了进度指示的意义。在发生阻塞式 I/O 之前，务必先刷新相关内容。
8. 将爬山算法的预算与事件循环的预算分开管理。在爬山算法执行过程中，一旦预算耗尽，应在该调用处立即处理，避免导致事件循环崩溃。当八次应用程序评估耗尽了 2 美元的预算后，爬山算法几乎没有剩余空间。事件循环会专门捕获来自爬山算法的 BudgetExhaustedError 错误，然后继续处理其余的应用程序。
try:
system_prompt = run_hill_climbing(system_prompt, lessons, meter)
run_state["decisions_since_last_improvement"] = 0
except BudgetExhaustedError:
print(f"[loop3] hill-climbing skipped — budget exhausted at ${meter.spent:.4f}")
run_state["decisions_since_last_improvement"] = 0
在其他场景应用该模式
四循环设计并非仅限于保险领域。若要对其进行调整，可替换以下部分：
- 循环1的工具。此处为：查找、规则、信号、决策。而您需要的是适用于您所在领域的检索与操作工具。
- 真实标准。此处为：历史决策文件。而您需要的是自己标注好的数据集。
- 评判标准。此处为：字符串比较。而您需要的是针对您任务的特定正确性测试。
- 经验总结。此处为：承保规则。而您需要的是来自您所在领域的错误模式。
- 爬坡提示。此处为：承保系统的提示语。而您需要的是您的代理系统的提示语。
- 预算限制。此处为：2.00美元。而您需要设定的是每次运行所需的成本。
有一个明确的限制：法官必须具有确定性，且独立于执行者。如果无法为你的任务定义确定的正确性测试，系统就无法可靠地自我改进。
尝试一下
设置及四个命令：
# 1. Install
cd underwriting_loop
pip install -e ".[dev]"
export ANTHROPIC_API_KEY=sk-ant-...
# 2. Run a single application (debug mode - cheapest)
python -m src.main --mode single --app-id APP-001
# 3. Drain the pending queue with event loop
python -m src.main --mode event-loop --once
# 4. Run standalone hill-climbing (5 rounds)
python -m src.main --mode improve --rounds 5
# 5. Run the full offline test suite (no API key needed)
pytest tests/ -v
爬山法的输出结果如下：
seed score: 44% (9 failures)
round 1: 56% KEPT (7 failures)
round 2: 56% REVERTED (score 56% <= best 56%)
round 3: 69% KEPT (5 failures)
round 4: 75% KEPT (4 failures)
round 5: 69% REVERTED (score 69% <= best 75%)
Final test evaluation...
spend: $1.8342
test score: 75%
gain (vs baseline): +25%
在训练过程中，初始提示词的得分仅为44%。经过五轮迭代后，最优提示词在保留的测试集上的得分达到了75%，这一提升幅度相当于在没有人工提示工程干预的自动提示词优化方式下的25个百分点提升。
循环机制的作用
循环工程并非某种框架，而是一种将智能体的可靠性分解为多个层次的方法，每个层次都有单一职责和单一反馈机制。这四个循环分别针对四种不同的故障模式：
- 循环1负责确保依据充分：智能体在做出决策前必须参考相关证据。
- 循环2用于处理单次运行故障：一条错误的推理路径不应导致整个应用永久失效。
- 循环3负责提升处理效率与抗故障能力：一个出问题的应用不应阻碍整个流程的运行。
- 循环4通过根据累积的错误重新编写提示词，来解决反复出现的错误问题。
将这些错误堆叠起来，加入确定的判断机制，再用预算控制和篡改检测来保护所有内容，如此一来，系统无需人工干预就能随着时间逐步得到显著提升。
你能获得的好处
应以审慎的态度采用这种设计。关于智能体循环的诸多讨论中最值得警惕的观点是Addy Osmani提出的“认知放弃”：即让循环完全掌控一切，不再查看其输出结果。尽管如此，其优势依然是真实且可量化的。
- 无需人工即可自我改进。通过爬山算法，训练准确率从44%提升到了75%，而且系统还能自行从错误模式中提取出正确的承保规则。
- 层与层之间的故障隔离。 单个格式错误的应用程序不会导致整个流程中断。Loop 3能够捕获每个应用程序的错误，将其记录下来后继续运行。
BudgetExhaustedError是唯一会上报到更高层的异常，这是有意为之。 - 可衡量且可审计的进展。
measure_gain()能够生成具体的数值，通过对比有状态与无状态场景的结果，展示改进效果。可以向相关方以百分比和成本金额的形式汇报改进情况。 - 防篡改的评估机制。 真实数据上的SHA-256校验和能够防止自优化系统中最常见的隐蔽故障，即有人篡改答案以使得分看起来更好。如果校验和不一致，则不会执行任何操作。
- 不断积累的知识。从失败中总结出的经验会持续保留，因此每次新的Loop 1调用都会在之前经验的基础上更明智地启动，无需重新从头学习。
- 生产者与消费者保持解耦。通过文件队列，生成待处理列表的模块与评估流程相互独立。可以暂停、恢复、查看或重放队列中的任务，而代理代码则不会受到影响。
- 人工介入是一种特殊状态。由
refer决策处理的结果会被存入human_queue.json，并视为已通过处理。对于需要人工决策的任务，循环不会再次尝试处理。 - 重试具有针对性。失败后的反馈会指出问题所在，例如“你选择的是‘修改’，但正确答案应是‘参考’，请重新审视风险因素”。这种方式比随机重采样能更快找到正确答案。
成本情况
- 令牌数量增长迅速。在16次训练任务中进行5轮爬山算法处理，至少需要80次Loop 1运行，再加上反思与审计操作。一次使用
--mode improve模式的运行会消耗2.00美元预算中的1.83美元。超出预算并非异常情况，而是此类运行通常的结局。 - 延迟问题确实存在。处理一个完整流程的应用程序可能需要30到60秒，而在事件循环模式下处理16个应用程序则需要数分钟。这种延迟适合批量处理，但不适合交互式使用。
- 非确定性会掩盖故障。循环1使用实时模型，因此同一应用程序中的相同提示可能会产生不同的工具序列。故障可能无法重现，而要有意义的准确率数据则需要多次评估运行。
- 初始化成本很高。在首次运行之前，需要运行状态、待处理队列、带有校验和的历史决策记录、最佳提示文件以及学习内容列表。这需要的基础设施规模比一次简单的
agent.invoke()调用要多出一个数量级。 - 过拟合问题始终存在。如果没有审计机制，爬山算法最终会生成一种记住训练答案的提示，在训练数据上能得到完美分数，但在测试数据上的表现则几乎靠运气。审计机制是防止模型自我提升变成自我欺骗的唯一手段。
- 认知上的放弃是一种真正的故障模式。循环越可靠，人们就越少查看其输出结果；而一旦停止阅读，就很难察觉到循环已开始以新的方式出现故障。无人监控的循环需要外部监督，因为循环本身无法自行检查语义的正确性。
- 调试涉及整个系统结构。错误的决策可能源于第4层循环中的错误处理、第2层循环中遗漏的规则、第1层循环中工具返回的空值，或是第3层循环中损坏的运行状态。要追踪四个嵌套循环中的故障，比追踪单个函数中的故障要困难得多。
真的有必要构建循环吗？
对公共循环结构的分析给出了明确建议：对于那些重复性高、可审核且价值大的任务，应设置内置检查机制；而对于其他任务，则只需触发一次即可。归根结底就是两个问题。
- 这项工作是否具有重复性，或者需要许多无人能预料的步骤？ 如果只需一次触发即可完成，那么使用循环只会增加开销且无法形成有效架构。
- 智能体能否自行判断任务是否已完成？ 若没有测试、预期格式或参考答案，循环就永远无法结束，只会无休止地继续。
两个问题的答案都必须是“是”。如果有任何一个答案是“否”，则应跳过循环。
适用场景
重复性高、处理量大的工作。 比如16份保险申请、500个代码文件、1000条支持工单。只有当任务被多次执行而非仅执行一次时，循环的成本才能得到弥补。
存在确定的完成检测机制。即真实的标签、测试套件、模式验证工具，或是二进制通过/失败规则的评分标准。评判者绝不应直接调用模型，因为基于模型的评判方式会引入自身的不可靠性。
智能体能够自行验证其工作结果。Anthropic的Agent SDK认为这是最容易被忽视的步骤。这里可通过is_correct()函数实现验证，在编程智能体中则由测试套件来完成。如果智能体无法检查自己的输出，就无法形成完整的反馈循环。
带标签的数据或可重复使用的成功标准。爬山法需要包含已知答案的训练集与测试集。没有这些数据，虽然可以分析失败原因，却无法判断这些分析是否真正有效。
没有人在监控每一次运行。 想想那些需要整夜处理的承保批次、定时进行的审计或文档处理流程，这些场景中根本没有人会参与内部循环，而这种设计正是为这类场景设计的。如果能够立即查看每个输出结果，直接进行提示会更快。
每次评估运行的成本是可以接受的。 一次爬山算法的运行成本相当于对所有训练样本进行完整处理的一次费用。如果这笔费用难以承担，那么Loop 4的成本也同样高，因此应减少运行频率或使用更小的训练集。
你需要数据支撑。 measure_gain() 出现的原因是“系统变好了”这一说法必须用具体数字来量化。如果利益相关方需要数据，循环工程就能提供这些数据；如果“看起来更好了”就足够了，那使用这个功能就显得多余了。
拟合效果不佳
一个提示词就能搞定。重新格式化CSV文件、从文档中提取日期、总结段落——这些任务的步骤数量都是可预知的，因此可以直接编写脚本或使用一次性提示词。在简单问题上使用循环只会增加架构上的复杂度。
“完成标准”纯属主观。“让这封邮件更专业”这类需求并没有明确的判断标准，再让另一个模型来评估美观度只会引入更多变量，却无法真正解决问题。应先明确目标，比如字数控制在100字以内、语气专业、不使用感叹号，之后再进行检查。
未经审核，操作将不可撤销。发送邮件、合并拉取请求、删除记录、转账——这些操作都需要严格限制权限，并且在执行删除、发送或支付之前必须暂停确认。如果允许循环执行不可撤销的操作而不设置审核机制，那问题就迟早会发生。
预算不足以覆盖评估费用。如果2.00美元都已紧张，那么在处理完所有训练任务之前，用于评估的资金就会被耗尽。系统会通过BudgetExhaustedError来处理这种情况，但对于预算有限的场景，真正的解决办法是使用更小的训练集、减少训练轮次，或者等到有足够资金后再进行第4轮训练。
不存在状态持久化功能。如果无法将运行状态、所学内容及任务队列保存到磁盘或数据库中，那么系统在崩溃后就无法继续运行，也无法在不同会话之间积累学习成果，更无法防止在运行过程中预算耗尽。无状态环境需要采用另一种架构。
你仍在探索中。使用合适的工具调用单个智能体，就能发现其出错之处，而嵌套循环只能显示整个系统的故障点。先对单个智能体进行测试，等熟悉了它的故障模式后再逐步增加层次。
最后一条经验法则来自Rhys Sullivan，他警告要避免两种极端：只为增加成本而运行的“tokenmaxxing”循环，以及过于简单的“promptchud”式单行指令。只有当循环能在真实的测试数据集中复现实际产品故障时才有意义，否则即便看起来很厉害也毫无价值。
有效循环的八个特性
循环通常会以一些并不理想的方式终止。它们不会停止运行，会耗尽预算，或者交出被标记为已完成但实际上是垃圾的成果。对公共循环结构的分析发现，每个能够稳定运行的循环都包含八个要素，而这些要素都对应着该代码库中的某个设计决策。
1. 可验证的目标
应定义“已完成”的标准，以便机器能够进行验证，而非为了让人类感觉满意。“做出良好的承保决策”这一标准是无法验证的。is_correct(decision, app_id)函数通过将当前决策与历史决策对比，得出匹配或不匹配的结果。如果没有可验证的目标，循环就永远不知道何时该停止，会将看似可靠但实际错误的输出当作已完成的工作交出来。该目标应是一个二元谓词，而非对质量的描述。训练集与测试集的划分则能确保模型不会仅依赖对过往数据的记忆。
2. 硬性终止条件
Max总会设置预算限制或时间限制，从未有过例外。切勿以为模型会自动停止运行。像shell中的while :; do ... done循环这样的基础结构在模型端并没有任何使其停止的机制，因此会一直运行直到有人终止进程；而OpenAI的SDK提供了max_turns参数，因为经验表明模型无法可靠地自行停止。该代码库在三个层面设置了停止机制：Loop 1中的MAX_STEPS = 8可防止工具链无限循环，Loop 2中的max_retries = 3可防止无限重试，而通过CostMeter设定的BUDGET_USD = 2.00则是最终的保障措施。当达到步数上限时，Loop 1会返回ran_out_of_steps；当重试次数用尽时，Loop 2会返回passed=False；而当预算耗尽时，错误会向上层传递并终止所有操作。
3. 优质工具
智能体可使用的操作必须可靠、描述清晰且范围明确。循环本身很简单，实际工作都在工具中完成。Anthropic的建议是，工具集及其文档必须经过精心设计，因为描述模糊会导致调用操作也模糊不清，而功能过强的工具则可能引发无法挽回的错误。这里有四种工具，每种负责一项任务：无副作用的纯检索、针对静态知识库的关键词搜索、模拟的外部数据获取，以及一种强制遵循指定结构模式（Literal["accept", "modify", "decline", "refer"]）的结构化终端操作，同时还需包含操作理由和明确标记。如果没有最后这些要素，智能体返回的文本将难以被准确解析。
4. 记忆
保留历史记录但进行总结，以避免上下文过于冗长；对于超过一次对话长度的内容，需将状态写入磁盘。Anthropic的long-running-agent框架能让智能体从git和进度文件中读取之前的状态，从而实现逐步推进，并为下一次迭代保持环境整洁。原始的上下文窗口在重启、崩溃或达到大小限制时会被丢失。此处有两种机制在不同时间尺度上发挥作用：在单次运行过程中，LangGraph的add_messages函数会为单个应用积累完整的工具对话内容；而在多次运行之间，run_state.json中的lessons列表则会积累所有应用及所有重启过程中的精简规则。distill_lesson()函数负责压缩处理：它存储的是可通用化的单一规则而非原始的故障记录，这样既能保持列表长度较短，也能让注入的上下文更易于管理。
5. 独立的检查机制
生成、判断、修正、重复，制作者绝不会给自己作品打分。Anthropic的SDK认为验证作品是最容易被忽视的步骤，但这一验证必须独立于生成过程。能够自我打分的智能体只会变成唯命是从的工具，而仅让另一个智能体负责打分就能避免这种情况。该代码库包含三个独立级别的检查：
- 评分标准检查（结构）。
RubricMiddleware使用纯布尔表达式对解析后的输出进行九项规则检测，无需依赖模型。 - 评判检查（正确性）。
is_correct()通过精确的字符串比对来检查答案是否与存储的标签一致。 - 审计检查（反作弊）。
audit_prompt()通过二进制模型分类来检测是否存在死记硬背的答案，而非进行质量评估。
评分标准与评分员完全由确定性因素决定。审核过程虽会使用模型，但仅用于二元结构问题（“其中是否包含应用ID？”），而这正是模型评分员最可靠的领域。
6. 复杂任务的规划方案
对于规模较大、需要多步骤完成的任务，应让智能体在行动前先制定计划；而对于小型任务则可省略此步骤。Anthropic的长期智能体研究显示，一次只处理一个功能至关重要，因为要求一次性解决所有问题的循环会导致系统混乱，而每次只进行一次修改则能更快收敛到正确结果。在这个代码库中，提示语按顺序列出了四个工具步骤，这实际上就构成了一种内置的规划方案：获取数据、获取规则、获取外部信号、最终做出决策。这样的顺序能防止智能体在不参考证据的情况下直接做出决策，而八步的上限则足以容纳完整的流程以及一次重试机会。
7. 日志记录
记录每一个想法、操作及结果，这样在凌晨3点进行调试时就有内容可查。如果某个步骤没有被记录，那么从调试的角度来看它就从未发生过。没有日志的话，嵌套循环中的故障几乎无法诊断：是哪个循环、哪个应用程序、哪次尝试、哪个工具返回了错误结果？三层结构可以解决这个问题。诸如 [loop3] 3/16 APP-009 ... 这样的进度行会实时更新，而在stdout被缓冲的Windows系统中这是不可或缺的。决策行会为每个应用程序用一行文字展示决策结果、状态及运行耗时。记录行则用于确认数据提取是否已完成。
8. 成本意识
循环会快速消耗代币，因此应从规模较小且有限的范围开始，逐步扩展，并在让循环持续运行之前先观察几次运行情况。在16个训练任务中进行一轮爬坡训练，再加上五轮反思与审计调用，费用几乎达到2.00美元。只有在了解每次迭代的成本后，才能设定预算并决定停止。CostMeter能精确到小数点后四位来记录支出，在费用超出预算之前会抛出BudgetExhaustedError异常：
# The ordering matters:
if self.spent >= self._budget: # check BEFORE adding
raise BudgetExhaustedError(...)
self.spent += cost # add AFTER the check clears
事件循环会单独检测爬坡过程中的预算耗尽情况，与处理过程中的耗尽情况分开处理，因此即使在自我提升阶段中途预算用尽，也不会影响队列中其余任务的执行。该循环会以优雅的方式降级：跳过成本较高的提升步骤，继续处理剩余任务。
自行运行前的检查清单
在让任何循环无人监控地运行之前，请确认以下几点：
- 目标是可以通过机器验证的（此处为使用
is_correct()函数与真实值对比）。 - 每个阶段都有明确的终止条件（
MAX_STEPS、max_retries、BUDGET_USD）。 - 相关工具都有明确的职责范围和文档说明（四种工具，各负责一项任务）。
- 状态会在重启后依然保留（
run_state.json文件以及课程列表）。 - 检查机制与生成机制是分离的（采用评分中间件以及确定的评判标准）。
[loop3]中的进度行）。CostMeter）。只要缺少其中任何一项，该循环在实际运行中就会失败，虽然不一定立即出现问题，但最终必然会失败。大多数实际运行故障都源于缺乏验证步骤、停止条件或持久化状态。先解决这些问题，再调整其他各项。
相关阅读
- 通过 FastAPI Webhooks 实现并行 LangGraph 智能体对拉取请求的审查 —— 讲解如何利用 Webhook、Redis 队列及四个 LangGraph 智能体，在约 30 秒内将拉取请求转化为带有评分的结构性 AI 审核结果，同时介绍如何使用 ngrok 进行本地测试。
- 利用 LangGraph 构建 LEPA 支持智能体 —— 介绍如何通过 Wire SupportState、intake 和 classify 节点、条件边以及 START/END 标记，让学校支持请求能够得到明确答复或自动跳转至相关文档。