企业官网建设流程全解析

1. 感知（Discover）：读取当前状态 2. 规划（Plan）：决定下一步动作 3. 执行（Execute）：调用工具或生成代码 4. 验证（Verify）：检查是否满足终止条件 → 不满足就回到第1步

class TurnBasedLoop: """ 最基础的Agent循环：人类在每一步检查并决定是否继续 """ def __init__(self, llm, tools: dict, max_turns: int = 10): self.llm = llm self.tools = tools self.max_turns = max_turns self.history = []
def run(self, task: str) -> str: self.history.append({"role": "user", "content": task}) for turn in range(self.max_turns): # Agent执行一步 response = self.llm.invoke(self.history) # 如果Agent调用了工具 if response.tool_calls: for tool_call in response.tool_calls: result = self._execute_tool(tool_call) self.history.append({ "role": "tool", "content": str(result) }) # 【关键】每一轮工具调用后暂停，等人类审查 human_feedback = input( f"[Turn {turn+1}] 工具执行完毕，结果：{result[:200]}\n" f"继续(c) / 修改指令(m) / 停止(q)：" ) if human_feedback == 'q': return response.content elif human_feedback == 'm': new_instruction = input("输入你的修改指令：") self.history.append({ "role": "user", "content": new_instruction }) else: # Agent没有调用工具，说明它认为任务完成了 return response.content return "达到最大轮次限制，任务可能未完成。" def _execute_tool(self, tool_call): tool_fn = self.tools.get(tool_call.name) if tool_fn: return tool_fn(**tool_call.args) return f"未知工具: {tool_call.name}"

适用场景：零散的短期任务、初次接触某个代码库时的探索性工作。

局限性太明显了——你得全程盯着。如果你的验收标准可以量化（比如"测试通过率100%"），那接下来这个模式更适合你。

模式二：目标循环（Goal Loop）—— 评审模型当刹车

这是目前最流行的模式。固定一个可验证的目标，每一轮结束后用一个独立的评审模型来判断是否达标。

class GoalLoop: """ 目标驱动循环：用独立评审模型判断任务是否完成 核心设计：执行者和评审者分离（Maker-Verifier Separation） """ def __init__( self, executor_llm, # 干活的模型（可以用强模型） evaluator_llm, # 评审的模型（用小模型就行，省钱） tools: dict, max_attempts: int = 5 ): self.executor = executor_llm self.evaluator = evaluator_llm self.tools = tools self.max_attempts = max_attempts
def run(self, task: str, success_criteria: str) -> dict: """ task: 要完成的任务描述 success_criteria: 机器可验证的成功标准 例如："所有单元测试通过" 或 "Lighthouse分数 > 90" """ context = {"task": task, "history": [], "attempts": 0} for attempt in range(self.max_attempts): context["attempts"] = attempt + 1 # 1. 执行者工作 execution_result = self._executor_step(context) context["history"].append({ "phase": "execution", "result": execution_result }) # 2. 评审者打分（关键：不是执行者自己判断！） evaluation = self._evaluate( task=task, criteria=success_criteria, execution_result=execution_result ) context["history"].append({ "phase": "evaluation", "score": evaluation["score"], "feedback": evaluation["feedback"] }) # 3. 达标了就停 if evaluation["passed"]: return { "status": "success", "attempts": attempt + 1, "final_result": execution_result, "score": evaluation["score"] } # 4. 没达标，把评审意见反馈给执行者 context["feedback"] = evaluation["feedback"] print(f"[Attempt {attempt+1}] 评分: {evaluation['score']}/100" f" | 反馈: {evaluation['feedback'][:100]}...") return { "status": "failed", "attempts": self.max_attempts, "reason": "达到最大尝试次数" } def _executor_step(self, context: dict) -> str: prompt = f"""任务：{context['task']}

当前进度：第{context[‘attempts’]}轮"“”

if context.get("feedback"): prompt += f"\n上一轮评审反馈：{context['feedback']}" prompt += "\n请根据反馈改进你的方案。" response = self.executor.invoke(prompt) # 执行工具调用 if response.tool_calls: for tc in response.tool_calls: tool_fn = self.tools.get(tc.name) if tool_fn: tool_fn(**tc.args) return response.content def _evaluate(self, task: str, criteria: str, execution_result: str) -> dict: eval_prompt = f"""你是一个严格的代码评审者。

任务目标：{task}
成功标准：{criteria}
实际执行结果：{execution_result}

请从以下维度打分（0-100）：

1. 功能完成度（40分）
2. 代码质量（30分）
3. 是否引入新问题（30分）

返回JSON格式：
{{"score": 分数, "passed": 是否达标(布尔值), "feedback": "改进建议"}}"""

response = self.evaluator.invoke(eval_prompt) # 解析评审结果 import json try: result = json.loads(response.content) result["passed"] = result["score"] >= 80 return result except json.JSONDecodeError: return {"score": 0, "passed": False, "feedback": "评审模型返回格式异常"}

关键设计：Maker-Verifier分离。

执行者永远不自己判断"我干完了"——这就像让写代码的人自己写考试评分标准，大概率给自己放水。用一个独立模型来当裁判，Loop才可靠。

适用场景：代码重构、性能优化、任何可以量化验收标准的任务。

模式三：定时循环（Timed Loop）—— 时钟当刹车

每隔固定时间触发一次。跟cron几乎一模一样，唯一区别是：cron跑的是固定脚本，Timed Loop跑的是一个能根据当前状态调整行为的AI工作流。

import schedule import time from datetime import datetime

class TimedLoop:
"""
定时循环：按固定间隔触发Agent工作流
适用场景：每日代码质量扫描、定时日志分析、周期性安全检测
"""
definit(self, llm, tools: dict, state_file: str = "loop_state.json"):
self.llm = llm
self.tools = tools
self.state_file = state_file
self.state = self._load_state()

def run_security_scan(self): """每天凌晨2点自动扫描依赖漏洞""" print(f"[{datetime.now()}] 开始定时安全扫描...") prompt = f"""你是一个安全运维Agent。

上次扫描时间：{self.state.get(‘last_scan’, ‘从未’)}
上次发现的未修复漏洞：{self.state.get(‘open_vulns’, 0)}

请执行以下操作：

1. 扫描项目依赖中的已知CVE漏洞

2. 与上次扫描结果对比，标记新增和已修复的漏洞

3. 对高危漏洞生成修复方案

4. 输出扫描报告"""

   response = self.llm.invoke(prompt) # 执行扫描工具 if response.tool_calls: for tc in response.tool_calls: tool_fn = self.tools.get(tc.name) if tool_fn: result = tool_fn(**tc.args) # 更新状态 self.state["last_scan"] = datetime.now().isoformat() self._save_state() print(f"[{datetime.now()}] 扫描完成")

   def run_log_analysis(self):
   """每小时分析一次系统日志"""
   prompt = """分析过去1小时的系统日志：

5. 统计错误频率Top5

6. 识别是否有异常模式（如突发错误激增）

7. 如果发现严重问题，生成告警"""
   # … 执行逻辑类似
   pass

   def _load_state(self) -> dict:
   import json
   try:
   with open(self.state_file, ‘r’) as f:
   return json.load(f)
   except FileNotFoundError:
   return {}

   def _save_state(self):
   import json
   with open(self.state_file, ‘w’) as f:
   json.dump(self.state, f, indent=2, ensure_ascii=False)

启动定时循环

ifname== "main":
# loop = TimedLoop(llm=my_llm, tools=security_tools)

# 每天凌晨2点跑安全扫描 schedule.every().day.at("02:00").do( lambda: loop.run_security_scan() ) # 每小时跑日志分析 schedule.every().hour.do( lambda: loop.run_log_analysis() ) print("定时循环已启动...") while True: schedule.run_pending() time.sleep(60)

注意看那个state_file——这是Timed Loop和普通cron的本质区别。Agent有记忆，它知道上次扫描发现了什么、哪些漏洞还没修，能做出增量判断而不是每次从零开始。

模式四：主动循环（Proactive Loop）—— 事件当刹车

最高级的模式。由事件触发（比如来了一个新的bug报告），全自动运行，不需要人看着。

class ProactiveLoop: """ 主动循环：事件驱动的无人值守Agent系统 适用场景：bug自动修复流水线、工单自动处理、PR自动审查 """ def __init__(self, llm, tools: dict, config: dict): self.llm = llm self.tools = tools self.config = config # 三道刹车 self.max_iterations = config.get("max_iterations", 20) self.max_budget_usd = config.get("max_budget", 5.0) self.no_progress_limit = config.get("no_progress_limit", 3)
async def handle_bug_report(self, bug: dict): """ 收到bug报告后自动触发修复流程 完全无人值守，从分析到提交PR """ print(f"[主动循环] 收到bug: {bug['title']}") context = { "bug": bug, "iteration": 0, "total_cost": 0.0, "files_touched": set(), "tests_passing": False, "history": [] } while context["iteration"] < self.max_iterations: context["iteration"] += 1 # 检查预算上限 if context["total_cost"] > self.max_budget_usd: print(f"[停止] 超出预算: ${context['total_cost']:.2f}") break # 检查无进展 if self._detect_no_progress(context): print(f"[停止] 连续{self.no_progress_limit}轮无实质进展") break # 1. Agent分析并修复 action = await self._agent_step(context) context["history"].append(action) context["total_cost"] += action.get("cost", 0) # 2. 记录修改的文件 for f in action.get("files_modified", []): context["files_touched"].add(f) # 3. 跑测试验证 test_result = await self._run_tests() context["tests_passing"] = test_result["all_passed"] if test_result["all_passed"]: # 4. 测试通过 → 提交PR pr_url = await self._create_pull_request(context) return { "status": "fixed", "pr_url": pr_url, "iterations": context["iteration"], "cost": context["total_cost"] } print(f"[迭代{context['iteration']}] 测试未通过，" f"继续修复... 累计费用: ${context['total_cost']:.2f}") return {"status": "failed", "reason": "达到循环上限"} def _detect_no_progress(self, context: dict) -> bool: """检测Agent是否在原地打转""" if len(context["history"]) < self.no_progress_limit: return False # 检查最近N轮是否修改了相同的文件但没有新的测试通过 recent = context["history"][-self.no_progress_limit:] recent_files = [set(a.get("files_modified", [])) for a in recent] # 如果最近N轮修改的文件完全相同 if len(recent_files) >= 2 and all(f == recent_files[0] for f in recent_files): return True return False async def _agent_step(self, context: dict) -> dict: prompt = f"""Bug标题：{context['bug']['title']}

Bug描述：{context[‘bug’][‘description’]}
当前迭代：{context[‘iteration’]}
历史记录：{context[‘history’][-3:] if context[‘history’] else ‘无’}

请分析bug原因并尝试修复。"""
response = self.llm.invoke(prompt)
# 执行工具调用并返回动作记录
return {
"action": response.content,
"files_modified": [],
"cost": 0.02 # 估算本次调用成本
}

async def _run_tests(self) -> dict: # 实际项目中调用pytest等测试工具 return {"all_passed": False} async def _create_pull_request(self, context: dict) -> str: return "https://github.com/example/repo/pull/42"

三道刹车全在这了：完成条件（测试全绿）、硬上限（max_iterations + max_budget）、无进展检测（_detect_no_progress）。

一个反直觉的发现：代码质量居然影响Agent运行成本

SonarSource团队今年做了一项控制实验（660次Claude Code试验，6个代码仓库对），结论很有意思：

代码干不干净，不影响Agent能不能完成任务——但影响花多少钱。

什么意思？就是说你的代码库越乱，Agent来回翻文件的次数越多，浪费的token越多。

他们算了一笔账：一个团队每周跑500个Agent任务，每次$2，代码从"中等脏"优化到"比较干净"，一年能省大约$4160。

这个数字不算大，但考虑到它只是"代码整洁"这一个变量带来的收益，而且随着Agent使用频率增加效果会线性放大——还是有意义的。

对我们来说的实操建议：在Loop系统中引入一个PostToolUse钩子，每次Agent修改文件后自动跑lint和类型检查。不是为了让Agent产出更好的代码（反正它都能完成任务），而是为了防止代码库在Agent的反复修改下逐渐退化——今天干净的代码，不加约束，半年后就是Agent多花7-8%token的元凶。

这块经验在我参与黑箭科技多个项目的Agent化改造中都验证过，代码治理这件事，不是道德选择，是经济选择。

落地你的第一个生产级Loop系统

说了这么多模式，真正落地的时候需要注意什么？我总结几个核心构件。

构件清单

一个能稳定运行的Loop系统需要凑齐5个东西：

1. 自动化触发器 —— 定时/事件/目标，决定什么时候开始 2. Maker-Verifier分离 —— 干活的人和评分的人不能是同一个 3. 状态持久化 —— 让Agent记住上次做到哪了 4. 三道刹车 —— 完成条件 + 硬上限 + 无进展检测 5. 成本追踪 —— 每个Loop跑了多少token、花了多少钱

下面是一个把这些构件组装在一起的完整示例：

""" 生产级Loop系统骨架 整合了5个核心构件，可直接作为项目模板使用 """ import json import time from datetime import datetime from pathlib import Path

class ProductionLoopSystem:
"""生产级Agent循环系统"""

def __init__(self, config: dict): # 构件1：配置与触发 self.config = config self.trigger = config["trigger"] # "cron" | "event" | "goal" # 构件2：Maker-Verifier分离 self.executor = config["executor_model"] # 干活用强模型 self.verifier = config["verifier_model"] # 评审用小模型 # 构件3：状态持久化 self.state_path = Path(config.get("state_file", "loop_state.json")) self.state = self._load_state() # 构件4：三道刹车 self.max_iterations = config.get("max_iterations", 20) self.max_budget = config.get("max_budget_usd", 5.0) self.no_progress_limit = config.get("no_progress_limit", 3) # 构件5：成本追踪 self.cost_tracker = {"total_tokens": 0, "total_cost": 0.0, "runs": []} def run(self, task: str, criteria: str) -> dict: start_time = time.time() iteration = 0 file_history = [] while iteration < self.max_iterations: iteration += 1 # 刹车1：预算上限 if self.cost_tracker["total_cost"] > self.max_budget: return self._result("budget_exceeded", iteration) # 刹车2：无进展检测 if self._check_stall(file_history): return self._result("stalled", iteration) # 执行步骤 result, files_modified, token_count = self._execute_step( task, self.state.get("feedback", "") ) file_history.append(set(files_modified)) self.cost_tracker["total_tokens"] += token_count self.cost_tracker["total_cost"] += self._calc_cost(token_count) # 评审步骤（独立的Verifier） passed, score, feedback = self._verify(result, criteria) if passed: return self._result("success", iteration, score) # 把评审反馈写入状态，供下一轮使用 self.state["feedback"] = feedback self._save_state() print(f"[第{iteration}轮] 评分:{score} | 反馈:{feedback[:80]}...") return self._result("max_iterations", iteration) def _execute_step(self, task: str, feedback: str): """执行一步，返回(结果, 修改的文件列表, token数)""" # 实际项目中这里调用LLM API return ("执行结果", ["src/main.py"], 1500) def _verify(self, result: str, criteria: str): """独立评审，返回(是否通过, 分数, 反馈)""" # 实际项目中调用独立的Verifier模型 return (False, 65, "需要改进错误处理逻辑") def _check_stall(self, file_history: list) -> bool: if len(file_history) < self.no_progress_limit: return False recent = file_history[-self.no_progress_limit:] return all(f == recent[0] for f in recent) def _calc_cost(self, tokens: int) -> float: return tokens * 0.00001 # 简化计算 def _result(self, status: str, iterations: int, score: float = 0): return { "status": status, "iterations": iterations, "score": score, "total_cost": self.cost_tracker["total_cost"], "total_tokens": self.cost_tracker["total_tokens"], "timestamp": datetime.now().isoformat() } def _load_state(self) -> dict: try: return json.loads(self.state_path.read_text()) except FileNotFoundError: return {} def _save_state(self): self.state_path.write_text(json.dumps(self.state, ensure_ascii=False))

使用示例

ifname== "main":
config = {
"trigger": "goal",
"executor_model": "gpt-5.5", # 用强模型干活
"verifier_model": "gpt-5.4-mini", # 用小模型评审，省85%成本
"state_file": "project_state.json",
"max_iterations": 15,
"max_budget_usd": 3.0,
"no_progress_limit": 3
}

system = ProductionLoopSystem(config) result = system.run( task="为src/api/目录下所有函数添加JSDoc注释", criteria="npm run lint:jsdoc 退出码为0" ) print(json.dumps(result, indent=2, ensure_ascii=False))

从Prompt到Loop，到底变了什么

最后说点虚的，但我觉得挺重要。

2023年，核心能力是写好一个Prompt。
2024年，核心能力是管好Context——RAG、记忆、工具描述。
2025年，核心能力是搭好Harness——工具链、权限、反馈通道。
2026年，核心能力是设计Loop——触发逻辑、验证逻辑、终止逻辑。

每一层都在把上一层的核心技能从"竞争优势"降级为"基础设施默认值"。

2024年写一个好Prompt是稀缺能力，2026年连这都算不上门槛了。

但这不意味着Prompt没用了。就像有了微服务架构你还是要会写函数一样——Loop系统里每一个节点内部，仍然需要好的Prompt。只是你的工作重心从"写一个好Prompt"变成了"设计一个好的行为系统"。

这个转变有点像从单体架构到微服务——你不再调优一台服务器上的性能，转而关心编排、熔断、链路追踪。

工程能力在升级，不是被替代。

总结

4种Loop模式，核心区别在于谁来踩刹车：

落地5个核心构件，缺一不可：

1. 自动化触发器
2. Maker-Verifier分离
3. 状态持久化
4. 三道刹车（完成条件 + 硬上限 + 无进展检测）
5. 成本追踪

一个反直觉但很重要的认知：代码整洁度不影响Agent能不能完成任务，但影响花多少钱。把代码治理当成经济决策来做。

你目前在项目里用的是哪种Loop模式？踩过什么坑？欢迎评论区交流。

本文由黑箭科技技术团队原创，专注于AI工程化实践与技术分享。如有转载，请注明出处。