Loop Engineering 实战：用 LangGraph 构建三层嵌套反馈循环的 Multi-Agent 系统

副标题：从「一个 Agent 做所有事」到「三个 Agent 互相校验」——这才是 2026 年生产级 AI 系统的真实样子

引子：我犯了一个经典错误

上一篇文章里我讲了 Loop Engineering 的理论框架。

发出去之后，有读者问了一个很好的问题：

「你说要设计循环而不是提示 AI，但一个循环到底长什么样？能不能给我看一个真实的系统？」

这篇文章的目的就是回答这个问题。

但在给出答案之前，我要先坦白一个我自己踩过的坑，因为它很典型。

六个月前，我构建了第一个「Multi-Agent 系统」。架构图画得很漂亮，有 Planner、有 Executor、有 Reviewer。但实际运行时，这三个 Agent 其实是串行的、无反馈的：

text

Planner → Executor → Reviewer → 结束

这不是 Multi-Agent 系统。这是一个顺序调用了三次 LLM 的管道。

真正的 Multi-Agent 系统，和这个的区别，不在于你调用了几次模型，而在于：Agent 之间是否有真正的反馈回路，是否有基于状态的条件路由，是否有自我纠错的能力。

大多数 AI Agent 失败不是因为模型的问题，而是因为一个 Agent 试图做所有事情——规划、推理、工具选择、执行、记忆。在 Demo 里这说得通，但在生产中会崩溃。

这篇文章，我们从头构建一个真正有三层反馈循环的生产级系统。

一、我们要构建什么：一个 AI 代码审查系统

场景：你是一个有10名工程师的团队，每天有大量 Pull Request 需要 Review。人工 Review 是瓶颈。你想构建一个 AI Agent 系统，自动完成初步代码审查，并在发现严重问题时通知人工介入。

为什么这个场景适合 Multi-Agent + Loop Engineering？

因为代码审查本质上是一个多视角、多轮迭代的任务：

• 安全角度看：有没有注入漏洞、密钥暴露？
• 质量角度看：有没有死代码、重复逻辑？
• 测试角度看：覆盖率是否足够？
• 最终裁决：上述三个角度综合后，批准还是拒绝？

一个真实的生产系统要严格分离关注点。Planner Agent 将目标分解为任务 DAG（有向无环图），但它不执行任务。一个会执行任务的 Planner，本质上只是换了名字的单 Agent。

我们的系统有三层循环：

text

┌────────────────────────────────────────────────────────────┐
│  L3 外层：任务编排循环（有无新 PR？触发内层）                │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  L2 中层：多视角审查循环（三个专家 Agent 互相校验）    │  │
│  │  ┌────────────────────────────────────────────────┐  │  │
│  │  │  L1 内层：单视角修复循环（发现问题→自动建议修复）│  │  │
│  │  └────────────────────────────────────────────────┘  │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘

二、架构设计：在写代码之前先画清楚

在 Multi-Agent 系统中，所有 Agent 必须在实现开始之前就商定共享状态 Schema。这不是技术约束，而是组织治理要求。Schema 不匹配是多 Agent Review 中最常见的集成失败模式。

所以第一步，定义 State Schema。

Python

# state.py
from typing import TypedDict, Literal, Annotated
from langgraph.graph.message import add_messages
class ReviewState(TypedDict):
    # ── 输入 ──
    pr_diff: str                    # PR 的 diff 内容
    pr_title: str                   # PR 标题
    pr_author: str                  # 作者
    # ── 中间状态（三个专家 Agent 的输出）──
    security_findings: list[str]    # 安全审查结果
    quality_findings: list[str]     # 代码质量结果
    test_findings: list[str]        # 测试覆盖结果
    # ── 路由控制 ──
    security_score: int             # 0-10，越低越危险
    quality_score: int              # 0-10
    test_score: int                 # 0-10
    revision_count: int             # 当前已修订轮次
    max_revisions: int              # 最大修订轮次（成本控制）
    # ── 最终输出 ──
    final_verdict: Literal[
        "APPROVE",
        "REQUEST_CHANGES", 
        "NEEDS_HUMAN"               # 分数太低，需要人工接管
    ]
    final_summary: str
    # ── 记忆（跨循环持久化）──
    review_memory: str              # 历史审查模式记录

注意这里最重要的设计决策：

1. revision_count + max_revisions：这是成本控制的闸门。没有它，循环会永远运行。
2. NEEDS_HUMAN 状态：这是「人在循环中」（Human-in-the-Loop）的触发器。7企业多 Agent 治理要求：高风险决策需要人工审批检查点，比如批准一次关键代码合并。
3. review_memory：跨 Session 的持久化记忆，让系统从历史中学习。

三、L1 内层循环：单视角专家 Agent

让我们从最内层的循环开始。每个专家 Agent 遵循同样的模式：分析 → 发现问题 → 尝试给出修复建议 → 评分。

Python

# agents/security_agent.py
from anthropic import Anthropic
from state import ReviewState
client = Anthropic()
SECURITY_SYSTEM_PROMPT = """你是一位资深应用安全工程师。
你的工作边界：
- 只分析代码中的安全问题
- 不评价代码风格或架构
- 每个发现必须包含：严重程度(HIGH/MED/LOW)、描述、修复建议
- 评分标准：10分=无安全问题，0分=存在严重漏洞（如SQL注入、密钥暴露）
- 不要发散，不要评论其他方面
输出格式（严格遵守）：
FINDINGS:
- [HIGH] 描述 | 修复：...
- [MED] 描述 | 修复：...
SCORE: X/10
REASONING: 一句话解释评分理由"""
def run_security_review(state: ReviewState) -> dict:
    """
    L1 内层循环的核心：
    如果发现 HIGH 级别问题，自动触发一次修复建议生成
    """
    # 读取历史记忆（Context Engineering 层）
    memory_context = state.get("review_memory", "无历史记录")
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=2000,
        system=SECURITY_SYSTEM_PROMPT,
        messages=[{
            "role": "user",
            "content": f"""历史审查模式参考：
{memory_context[-500:]}
待审查 PR：{state['pr_title']}
作者：{state['pr_author']}
代码变更：
```diff
{state['pr_diff'][:4000]}
```"""
        }]
    )
    raw_output = response.content[0].text
    # 解析输出
    findings, score = parse_security_output(raw_output)
    # L1 内层循环：如果有 HIGH 级别问题，且未超过修订次数
    # 自动生成详细修复代码建议
    enhanced_findings = findings
    if any("[HIGH]" in f for f in findings):
        if state.get("revision_count", 0) < 2:  # 内层最多2轮
            enhanced_findings = enhance_with_fix_suggestions(
                findings, state["pr_diff"]
            )
    return {
        "security_findings": enhanced_findings,
        "security_score": score,
    }
def enhance_with_fix_suggestions(
    findings: list[str], 
    pr_diff: str
) -> list[str]:
    """L1 内层：针对 HIGH 级别问题生成具体修复代码"""
    response = client.messages.create(
        model="claude-haiku-4-5",  # 用更小的模型，节省成本
        max_tokens=1500,
        system="你是安全修复专家。只输出代码修复建议，简洁精确。",
        messages=[{
            "role": "user", 
            "content": f"""针对以下高危发现，给出具体修复代码：
{chr(10).join([f for f in findings if '[HIGH]' in f])}
原始代码上下文：
{pr_diff[:2000]}"""
        }]
    )
    fix_code = response.content[0].text
    return findings + [f"[FIX_SUGGESTION]\n{fix_code}"]

注意这里的设计：

内层循环（L1）只做一件事：发现 HIGH 问题，尝试给出修复建议，最多迭代2次。它不关心质量问题，不关心测试覆盖，关注点严格隔离。

四、L2 中层循环：Supervisor 编排多视角校验

这是整个系统的核心。LangGraph 支持三种核心 Multi-Agent 模式：Supervisor（一个编排者委托给子 Agent）、层级式（嵌套 Supervisor）和协作式（对等 Agent 共享消息队列）。

我们使用 层级式 + 条件边 构建中层循环：

Python

# graph/review_graph.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from state import ReviewState
from agents.security_agent import run_security_review
from agents.quality_agent import run_quality_review
from agents.test_agent import run_test_review
def route_after_review(state: ReviewState) -> str:
    """
    L2 中层循环的路由逻辑：
    这是整个系统最关键的函数——决定下一步做什么
    """
    security_score = state.get("security_score", 10)
    quality_score = state.get("quality_score", 10)
    test_score = state.get("test_score", 10)
    revision_count = state.get("revision_count", 0)
    max_revisions = state.get("max_revisions", 3)
    # 规则1：安全问题是一票否决
    if security_score < 4:
        if revision_count < max_revisions:
            # 还有修订机会：触发安全重审循环
            return "re_review_security"
        else:
            # 已达最大修订次数：升级到人工审查
            return "escalate_to_human"
    # 规则2：质量和测试都需要达标
    if quality_score < 6 or test_score < 6:
        if revision_count < max_revisions:
            return "re_review_quality_and_test"
        else:
            return "request_changes"
    # 规则3：所有分数达标
    if security_score >= 7 and quality_score >= 7 and test_score >= 7:
        return "approve"
    # 规则4：介于中间——需要人工判断
    return "escalate_to_human"
def increment_revision(state: ReviewState) -> dict:
    """每次进入重审循环，计数器+1（成本控制的核心）"""
    return {"revision_count": state.get("revision_count", 0) + 1}
def make_final_decision(state: ReviewState) -> dict:
    """综合三个 Agent 的结果，生成最终审查报告"""
    all_findings = (
        state.get("security_findings", []) +
        state.get("quality_findings", []) +
        state.get("test_findings", [])
    )
    avg_score = (
        state.get("security_score", 0) +
        state.get("quality_score", 0) +
        state.get("test_score", 0)
    ) / 3
    if avg_score >= 7:
        verdict = "APPROVE"
    elif avg_score >= 5:
        verdict = "REQUEST_CHANGES"
    else:
        verdict = "NEEDS_HUMAN"
    # 生成结构化审查报告
    summary = f"""## AI 代码审查报告
**PR**: {state['pr_title']}
**作者**: {state['pr_author']}
**审查轮次**: {state.get('revision_count', 0) + 1}
### 评分
- 安全性: {state.get('security_score', 'N/A')}/10
- 代码质量: {state.get('quality_score', 'N/A')}/10  
- 测试覆盖: {state.get('test_score', 'N/A')}/10
- **综合评分: {avg_score:.1f}/10**
### 主要发现
{chr(10).join(f'- {f}' for f in all_findings[:10])}
### 决定: **{verdict}**"""
    return {
        "final_verdict": verdict,
        "final_summary": summary
    }
def update_review_memory(state: ReviewState) -> dict:
    """
    L2 循环结束后：更新持久化记忆
    这是让系统随时间变得更聪明的关键
    """
    new_memory_entry = f"""
## 审查记录 {state['pr_title']}
- 安全评分: {state.get('security_score')}/10
- 质量评分: {state.get('quality_score')}/10
- 测试评分: {state.get('test_score')}/10
- 结果: {state.get('final_verdict')}
- 修订次数: {state.get('revision_count', 0)}
"""
    existing_memory = state.get("review_memory", "")
    # 只保留最近5条记录（避免上下文窗口爆炸）
    recent_entries = existing_memory.split("## 审查记录")[-5:]
    updated_memory = "## 审查记录".join(recent_entries) + new_memory_entry
    return {"review_memory": updated_memory}
# ── 构建图 ──
def build_review_graph(checkpointer):
    graph = StateGraph(ReviewState)
    # 添加节点
    graph.add_node("security_review", run_security_review)
    graph.add_node("quality_review", run_quality_review)
    graph.add_node("test_review", run_test_review)
    graph.add_node("increment_revision", increment_revision)
    graph.add_node("make_final_decision", make_final_decision)
    graph.add_node("update_memory", update_review_memory)
    # 入口：并行执行三个审查（Fan-out）
    graph.set_entry_point("security_review")
    graph.add_edge("security_review", "quality_review")
    graph.add_edge("quality_review", "test_review")
    # 核心：条件边——L2 中层循环的路由
    graph.add_conditional_edges(
        "test_review",
        route_after_review,
        {
            "re_review_security": "increment_revision",
            "re_review_quality_and_test": "increment_revision",
            "approve": "make_final_decision",
            "request_changes": "make_final_decision",
            "escalate_to_human": "make_final_decision",
        }
    )
    # 重审循环：回到对应的审查节点
    graph.add_conditional_edges(
        "increment_revision",
        lambda s: (
            "security_review" 
            if s.get("security_score", 10) < 4 
            else "quality_review"
        ),
        {
            "security_review": "security_review",
            "quality_review": "quality_review"
        }
    )
    # 收尾：更新记忆，结束
    graph.add_edge("make_final_decision", "update_memory")
    graph.add_edge("update_memory", END)
    return graph.compile(checkpointer=checkpointer)

五、L3 外层循环：生产级任务编排

现在把整个系统放入最外层的生产循环。

LangGraph 用于构建需要复杂分支逻辑、循环、持久化和人工介入控制的有状态多 Agent AI 工作流。它为 Klarna（8500万用户）、Uber、LinkedIn 和 Coinbase 的生产系统提供支持。

Python

# main.py：L3 外层循环
import asyncio
import time
from langgraph.checkpoint.postgres import PostgresSaver
from graph.review_graph import build_review_graph
from integrations.github import fetch_pending_prs, post_review_comment
# 生产配置
POSTGRES_CONN = "postgresql://user:pass@localhost/langgraph"
MAX_CONCURRENT_REVIEWS = 3     # 最大并发审查数
POLL_INTERVAL_SECONDS = 60     # 每60秒检查一次新 PR
COST_BUDGET_PER_PR = 0.50      # 每个 PR 最多花 $0.50
async def process_single_pr(pr: dict, graph, semaphore: asyncio.Semaphore):
    """处理单个 PR 的完整审查流程"""
    async with semaphore:  # 并发控制
        thread_id = f"pr-{pr['id']}-{int(time.time())}"
        initial_state = {
            "pr_diff": pr["diff"],
            "pr_title": pr["title"],
            "pr_author": pr["author"],
            "revision_count": 0,
            "max_revisions": 3,
            "review_memory": load_team_memory(pr["repo"]),  # 加载团队记忆
        }
        config = {
            "configurable": {"thread_id": thread_id},
            # 人工介入点：当 verdict 是 NEEDS_HUMAN 时自动暂停
            "interrupt_before": ["escalate_to_human_node"]  
        }
        try:
            final_state = await graph.ainvoke(initial_state, config)
            # 把审查结果发布回 GitHub PR
            await post_review_comment(
                pr_id=pr["id"],
                comment=final_state["final_summary"],
                verdict=final_state["final_verdict"]
            )
            print(f"✅ PR #{pr['id']} 审查完成: {final_state['final_verdict']}")
        except Exception as e:
            print(f"❌ PR #{pr['id']} 审查失败: {e}")
            # 失败不中断整个循环，记录日志继续下一个
async def run_outer_loop():
    """
    L3 外层循环：
    持续监控 GitHub，发现新 PR 就触发审查
    """
    # 生产级 Checkpointer：PostgreSQL（不是内存）
    # 原因：服务重启后可以恢复未完成的审查
    checkpointer = PostgresSaver.from_conn_string(POSTGRES_CONN)
    checkpointer.setup()
    graph = build_review_graph(checkpointer)
    semaphore = asyncio.Semaphore(MAX_CONCURRENT_REVIEWS)
    print("🚀 代码审查循环启动...")
    while True:  # L3 外层：永久运行
        try:
            # 读取当前状态：有哪些未审查的 PR？
            pending_prs = await fetch_pending_prs()
            if not pending_prs:
                print(f"💤 暂无待审查 PR，{POLL_INTERVAL_SECONDS}秒后重新检查")
                await asyncio.sleep(POLL_INTERVAL_SECONDS)
                continue
            print(f"📋 发现 {len(pending_prs)} 个待审查 PR")
            # 并发触发 L2 中层循环（每个 PR 一个独立图实例）
            tasks = [
                process_single_pr(pr, graph, semaphore)
                for pr in pending_prs
            ]
            await asyncio.gather(*tasks, return_exceptions=True)
        except KeyboardInterrupt:
            print("👋 收到停止信号，优雅退出...")
            break
        except Exception as e:
            # 外层循环永不崩溃：记录错误，继续运行
            print(f"⚠️ 外层循环错误（已记录，继续运行）: {e}")
            await asyncio.sleep(30)
if __name__ == "__main__":
    asyncio.run(run_outer_loop())

六、生产四大支柱：不做这些，别上线

企业 LangGraph 部署需要四个生产支柱：检查点存储到 PostgreSQL、LangSmith 追踪用于可观测性、基于中断的人工介入用于治理，以及部署目标（LangGraph Platform 或自托管容器化服务）。

支柱1：PostgreSQL Checkpointing（不是 MemorySaver）

Python

# ❌ 开发环境用这个
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()  # 服务重启，所有状态消失
# ✅ 生产环境必须用这个
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langgraph"
)

用 Postgres checkpointing 和连接池——可以防止在高并发下会遇到的 SQLite 锁争用问题。

支柱2：每个 Supervisor 循环必须有熔断器

对每个 Supervisor 循环设置最大修订次数限制——防止无声地燃烧你 API 预算的无限循环。

Python

# 在 route_after_review 里永远先检查这个：
if revision_count >= max_revisions:
    return "escalate_to_human"  # 强制退出循环

支柱3：Pydantic 验证 State（捕获90%的路由 Bug）

用 Pydantic 验证 State 和 Literal 类型做路由——在开发阶段就能捕获 90% 的路由 Bug。

Python

from pydantic import BaseModel
from typing import Literal
class ReviewStateV2(BaseModel):
    # Literal 类型让路由函数的返回值在运行前就被类型检查器验证
    final_verdict: Literal["APPROVE", "REQUEST_CHANGES", "NEEDS_HUMAN"] = "REQUEST_CHANGES"
    security_score: int = Field(ge=0, le=10)  # 强制在 0-10 范围内

支柱4：治理框架嵌入图结构，而不是事后添加

企业多 Agent 治理需要：每个 Agent 的基于角色的权限边界（最小权限）、每次 Agent 动作的决策审计日志、高风险决策的人工介入检查点，以及防止 Prompt 注入的输入/输出验证。治理框架应该内嵌到图架构中，而不是事后加上去。

七、三层循环的成本估算（真实数据）

这是很多教程不会告诉你的部分：这个系统到底花多少钱？

以一个中等规模的 PR（约500行 diff）为例：

如果你的团队每天有20个 PR，月成本约 $144。

相比之下，一个工程师花30分钟 Review 一个 PR，按时薪 $100 计算，月成本是 $10,000+。

但这里有一个关键的诚实声明：AI 审查不能完全替代人工审查。它最大的价值是：

1. 捕获明显的安全漏洞和代码质量问题（这两类问题占人工 Review 时间的 ~60%）
2. 给工程师提供第一轮 Review 摘要，大幅减少审查时间
3. 在 NEEDS_HUMAN 的情况下，确保人工 Review 聚焦在真正需要判断的问题上

八、我踩过的三个深坑

坑1：State 在并发时被竟态条件污染

当多个 PR 同时在审查时，如果 review_memory 字段同时被多个图实例写入，会产生数据竞争。

解法：每个 PR 使用独立的 thread_id，PostgreSQL checkpointer 会自动隔离不同线程的状态。这是用 PostgreSQL 而不是内存存储的核心原因之一。

坑2：Agent 输出格式不稳定导致 parse 失败

即使给了严格的输出格式约束，LLM 偶尔还是会输出不符合预期的格式，导致解析函数崩溃，整个图中断。

解法：所有解析函数必须有 try/except 保底，返回默认值而不是抛出异常。图的节点失败不应该中断整个外层循环。

坑3：上下文窗口随循环次数增长爆炸

每次重审循环，review_memory 会累积。跑到第15个 PR 时，上下文已经超过了 200K tokens。

Vault（或记忆文件）只有当 AI Agent 接管日常维护时，才能真正成为第二大脑。 但接管的前提是你要控制它的大小。

解法：记忆窗口滑动——只保留最近5条审查记录，旧记录归档到独立的 archive/ 文件而不是塞进上下文。

九、选择 LangGraph 而不是其他框架的 DECISIONS.md

这是很多文章不写，但我认为最有价值的部分。

当你需要对执行流程的显式控制和生产级持久化时，使用 LangGraph。

具体到我们这个场景的决策：

Markdown

# DECISIONS.md
## 决策：使用 LangGraph 而非 CrewAI
**时间**：2026年3月
**背景**：需要构建一个有嵌套循环的多 Agent 审查系统
**选择 LangGraph 的原因**：
1. 条件边（conditional_edges）支持基于状态的动态路由
   CrewAI 的角色分配是静态的，无法在运行时根据评分决定
   下一步交给哪个 Agent
2. PostgreSQL checkpointing 原生支持
   审查流程可能需要几分钟，需要在服务重启后恢复状态
3. interrupt_before 支持人工介入节点
   当安全分 < 4 时需要通知人工，这是治理要求
**不选 CrewAI 的原因**：
- CrewAI 适合角色明确、流程固定的多 Agent 场景
- 我们的循环次数和路由逻辑是动态的，不适合 CrewAI
**不选 AutoGen 的原因**：
- AutoGen 适合研究性和协作编码 Agent
- 我们需要生产级持久化，AutoGen 在这方面相对薄弱
**成本**：每个 PR ~$0.24，月预算上限设定为 $500
**回顾时间**：2026年6月（检查是否需要调整模型或策略）

结语：代码在 GitHub，架构在你脑子里

解决方案不是一个更好的提示词。解决方案是架构。

这篇文章给了你代码，但我更想给你的是一套架构思维：

• L1 内层：单一职责，可测量的验证条件，有限的修订次数
• L2 中层：严格的关注点分离，基于状态的条件路由，人工介入节点
• L3 外层：永不崩溃，优雅降级，成本受控

这三层不是 LangGraph 的专属模式。无论你用什么框架，只要你在构建需要自我校正的 AI 系统，这三层的思维模型都适用。

工具会变，架构思维不会。