Agent 能根据观察调整下一步行动，但自由选择并不等于流程可靠。生成代码、执行测试、分析错误和重新修改之间存在反馈环；多个 Agent 协作还会增加状态合并、交接和资源管理的复杂度。Loop Engineering、Graph Engineering 与 Harness Engineering分别控制循环、表达状态转换，并验证真实执行结果。

完整系统需要同时回答：当前在哪个状态，哪些操作已经完成，为什么进入下一节点，失败后怎样恢复，以及如何验证任务成功。Prompt 提供任务要求，运行时则把这些要求落实为路由、预算、持久化和测试。

Graph 结构

DAG是有向无环图，适合表达具有依赖关系的固定任务。带环的有向图允许边返回已执行过的节点，能够直接表达“生成—测试—修正”的反馈过程。

DAG 与带反馈环的状态图

DAG 调度器也可以为某个节点设置重试，不能把 DAG 理解成只能在全部步骤成功时运行。区别在于：任务重试属于执行策略，而回到另一个业务节点、更新状态并重新决策，需要显式控制流程。简单链路可以保持 DAG，只有需要反馈的部分才引入环。

状态图的三个组成部分是 State、Node 和 Transition：

组成 职责 设计重点
State 保存业务数据与运行进展 字段语义、版本、合并规则
Node 读取状态并完成有限操作 单一职责、输入输出、超时
Transition 决定下一步执行位置 条件完整、目标合法、失败出口

节点通常只返回发生变化的字段，即 Partial State Update。框架根据相应 Reducer 合并，而不是让每个节点自行复制并修改全部历史。

节点分工

把模型调用、检索、写入和校验全部放进一个 run_agent()，会使测试、权限和故障定位纠缠。可以按职责分成四类节点。

节点 工作 典型结果
LLM Node 分类、规划、生成候选代码 意图、方案、Patch
Data Node 读取文档、AST、向量库和业务数据 证据、符号、对象状态
Action Node 应用补丁、运行工具或提交业务变化 执行状态与副作用记录
Guard Node 检查权限、参数、预算和敏感数据 允许、拒绝、待审批

LLM Node 可以访问模型服务，因此仍涉及网络和计算成本；“不直接修改业务对象”比“完全没有副作用”更准确。Data Node 也需要限制读取范围，Action Node 需要明确影响目标，Guard 则应由确定性规则处理不可绕过的条件。

节点接口越清楚，越容易注入模拟依赖、设置独立超时和定位失败。Graph 负责协调这些能力，而不是替代各能力内部的验证。

条件路由

Conditional Edge读取状态，返回有限路由键，再映射到已注册节点。模型可以给出候选意图，但不能随意生成一个节点名并跳过必需的校验。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from typing import Literal, TypedDict


class RepairState(TypedDict):
    """保存修复任务的状态和有限重试进度。"""

    code: str
    test_status: Literal["pending", "passed", "failed", "error"]
    error_log: str
    attempts: int
    max_attempts: int


def route_after_test(state: RepairState) -> str:
    """先确认成功，再根据尝试次数与错误类型选择出口。"""
    if state["test_status"] == "passed":
        return "complete"
    if state["attempts"] >= state["max_attempts"]:
        return "human_review"
    if state["test_status"] == "failed":
        return "repair"
    return "stop"

每个返回值都应出现在 Path Map 中。未知状态应进入明确的停止或诊断分支，不能默默当成成功，也不能返回 None 后把错误留给后续步骤。

路由函数宜保持简单、可测试，不在判断过程中调用数据库或产生写入。需要外部资料时，先由节点取得数据，再依据状态路由。

循环边界

循环需要步数、时间、费用和进展四类限制。recursion_limit 限制一次图执行的计算步数，不是某个函数的超时，也不能终止已经卡住的外部进程。

1
2
3
4
5
6
7
8
9
10
from langgraph.errors import GraphRecursionError


def invoke_with_limit(graph, inputs: dict, config: dict) -> dict:
    """为同步图设置步数上限，保留达到上限的明确状态。"""
    bounded_config = {**config, "recursion_limit": 25}
    try:
        return {"status": "completed", "result": graph.invoke(inputs, bounded_config)}
    except GraphRecursionError:
        return {"status": "step_limit", "result": None}

这里的 25 是示例预算，需要依据流程长度设置。节点本身还需超时，长工具任务应支持取消，并确认子进程或容器真的结束。

Action Fingerprint根据工具名和规范化参数识别重复动作。相同错误参数反复出现而没有新观察时，可以提前转入降级或人工处理。但合法轮询也会重复调用，所以不能只看到相同哈希就认为死循环。

1
2
3
4
5
6
7
8
9
10
11
12
13
import hashlib
import json


def action_fingerprint(tool_name: str, arguments: dict) -> str:
    """把等价参数映射到稳定指纹，避免字典顺序影响判断。"""
    payload = json.dumps(
        {"tool": tool_name, "arguments": arguments},
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

指纹计数还可以结合目标版本、最新观察摘要和持续时间。多 Agent 交接需要额外检查 A→B→A 的往返是否产生进展，避免仅限制单个 Agent 后仍形成全局无限循环。

终止条件

显式 END 表达运行终点，但存在终点不等于一定能到达终点。要说明有限终止，需要保证每条路线有出口，循环消耗有限预算，节点能够完成或被超时终止。

可以用剩余尝试次数建立排名函数：

V(s)=max_attempts−attempts(s).V(s)=\mathrm{max\_attempts}-\mathrm{attempts}(s).

若每次返回修复环都使 VV 减少，且 V=0V=0 时不再回到修复节点，则这一重试环具有有限次数。对于多个嵌套循环，还需要全局预算或更完整的排名函数。

到达 END 可以是成功、失败、取消或待外部处理，状态中应保存原因。它不自动保证所有句柄都已回收，也不证明任务满足业务要求；执行器仍需清理连接、临时文件和容器，并验证最终产物。

状态归约

Reducer决定某字段如何处理节点返回的更新。普通覆盖适合当前状态，列表追加适合事件记录，按标识合并适合多个分支的产物。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from operator import add
from typing import Annotated, TypedDict


def merge_artifacts(old: dict[str, str], new: dict[str, str]) -> dict[str, str]:
    """合并互不冲突的产物，拒绝同一标识出现不同内容。"""
    result = dict(old)
    for artifact_id, content in new.items():
        if artifact_id in result and result[artifact_id] != content:
            raise ValueError(f"Conflicting artifact: {artifact_id}")
        result[artifact_id] = content
    return result


class SharedState(TypedDict):
    """分别声明追加、按键合并和单值覆盖的字段。"""

    messages: Annotated[list[dict], add]
    artifacts: Annotated[dict[str, str], merge_artifacts]
    current_status: str

add 对列表执行 old + new。因此返回 {"messages": []} 不会清空旧消息，只会追加空列表。消息更新还可以采用按消息 ID 合并的规则，避免修改同一消息时产生重复。

并发节点更新未配置归约的同一字段时，LangGraph 可能报告冲突，不能笼统解释为任何情况下都由最后完成者覆盖。对并发结果要明确合并与排序规则；列表追加也不自动成为具有幂等、交换律和冲突解决的 CRDT。

重试与降级

Retry Policy适合临时网络故障、限流或短暂不可用。指数退避可以表达为：

Δtj=min⁡(Δt0bj,Δtmax⁡),\Delta t_j=\min(\Delta t_0 b^j,\Delta t_{\max}),

并可加入抖动，避免多个请求同时再次访问。参数错误、权限拒绝和确定性业务失败不应无限重试。

代码测试失败通常应转换成状态中的 Observation，再由条件边回到修复节点；模型服务短暂限流则可以由节点级 Retry Policy 处理。二者分别是业务反馈环和调用重试。

重试没有成功时不更新共享状态，并不能证明外部副作用没有发生。请求可能已完成写入，却在回传时超时，因此写操作仍需幂等标识、状态查询和补偿机制。

Checkpoint

长任务可能跨越 HTTP 超时、服务重启和人工等待。Checkpointer保存运行状态与后续位置，使任务不必依赖一个长期存活的请求或内存对象。

需要保存的内容包括业务输入、节点结果、待执行动作、状态版本以及与外部操作的关联标识。检查点通常按运行引擎的步骤边界保存，不应假定每一个内部函数调用都有一个独立快照。

后端 适用情况 需要考虑
内存保存器 单进程测试和演示 重启后丢失
SQLite 本地调试、小规模服务 文件锁、连接方式、并发写入
PostgreSQL 多实例服务与长期状态 连接池、迁移、事务和备份
Redis 高频或短生命周期状态 持久化配置、TTL 和淘汰策略

SQLite、PostgreSQL、Redis 的保存器通常来自相应集成包，需要固定依赖并按要求初始化。选用了数据库，并不自动完成多租户授权、线程安全或全局业务事务。

会话标识

thread_id 标识一条状态历史，checkpoint_id 定位其中的具体版本，checkpoint_ns 区分嵌套图的检查点命名空间。它们用于定位状态，不是访问权限凭证。

1
2
3
4
5
6
7
8
9
config = {
    "configurable": {
        "thread_id": "tenant-a:repair-task-42"
    }
}

# 从已验证归属的历史快照取得配置，而不是自行猜测 checkpoint ID。
snapshot = graph.get_state(config)
selected_config = snapshot.config

用户输入的 HTTP Header 不能未经核对就用于读取任意会话。入口应从受信身份取得租户，验证任务归属，再构造 Config。checkpoint_ns 通常由图运行时管理，不能把手工填一个命名空间当成完整隔离实现。

持久化需要 TTL、保留期限、PII 脱敏、Schema 版本与迁移。凭证、数据库连接和沙箱客户端宜放在运行依赖中，不直接序列化进历史状态。

中断恢复

静态断点可在编译时通过 interrupt_before 或 interrupt_after 放到指定节点边界，适合调试固定位置。动态 interrupt() 则在节点内依据具体状态挂起，适合业务审批。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langgraph.types import Command, interrupt


def approval_node(state: dict) -> dict:
    """提交具体补丁与版本给审批入口，恢复后验证决定类型。"""
    decision = interrupt({
        "action": "apply_patch",
        "patch_digest": state["patch_digest"],
        "base_commit": state["base_commit"],
    })
    if not isinstance(decision, dict) or type(decision.get("approved")) is not bool:
        raise ValueError("An explicit approval decision is required")
    return {"approved": decision["approved"]}


# 首次运行由实际任务入口发起，获得中断结果后才进入审批流程。
# 同一任务完成身份、参数和版本核对后，再使用原配置恢复。
resume_input = Command(resume={"approved": True})

Command(resume=...) 由受信审批入口提交，不能把模型自己的“批准”文字作为授权。恢复时中断所在节点可能从开头重跑，所以中断前应避免不可重复副作用，或提供幂等保护。

持久化挂起可以释放正在等待的执行资源，但状态存储、队列和监控仍占用资源。客户端断开也不一定表示任务取消，需要独立任务状态与取消操作。

历史与分支

Time Travel从历史 Checkpoint 继续执行，Fork则在历史位置注入不同状态，形成另一条后续路线。它们用于调试、策略比较和失败分析。

1
2
3
4
5
6
7
8
9
10
11
12
def fork_with_strategy(graph, config: dict, strategy: str) -> dict:
    """从已验证历史点创建新检查点，并使用返回的配置继续。"""
    history = list(graph.get_state_history(config))
    before_generation = next(
        snapshot for snapshot in history
        if snapshot.next == ("generate",)
    )
    fork_config = graph.update_state(
        before_generation.config,
        {"strategy": strategy},
    )
    return graph.invoke(None, fork_config)

示例针对同步图；异步节点应使用相应异步接口。update_state 创建新的状态版本，原历史仍可保留，不能把它解释成擦除旧轨迹。分支可以存在于同一 thread 的检查点树中，并非必须更换 thread_id。

更换 thread ID 本身不会复制完整执行位置、待办与历史，跨 thread 迁移需要显式处理。What-If 实验还应固定 Base Commit、依赖、输入和测试环境，并把写操作限制在独立工作区或 Mock 中。

状态回放不等于外部世界回滚。 重放后的节点可能再次调用模型、运行工具或写入业务系统；真实副作用必须单独隔离、去重或补偿。

人工编辑

人工可以修改草稿、参数或 Patch，再恢复任务。update_state() 会按相应 Reducer 应用更新，as_node 说明该更新应视为哪个节点产生，从而影响后续位置。

1
2
3
4
5
6
7
8
updated_config = graph.update_state(
    selected_config,
    {"generated_code": corrected_code, "approved": False},
    as_node="generate",
)

# 人工修改后从生成节点的后继继续，重新运行验证。
result = graph.invoke(None, updated_config)

补丁被修改后，旧测试报告和旧审批应失效。不能用 as_node 假装安全审查已经完成，从而跳过真正校验。审计记录应保存操作者、修改前后摘要、原因和父检查点。

长期工作流

LangGraph 适合 AI 状态与反馈环，LlamaIndex Workflows 以事件驱动方式组织检索任务，Temporal/Cadence 适合跨服务、定时等待和持久调度。它们可以结合，而不是必须选择一个框架承包所有层次。

将模型或图运行封装为 Temporal Activity，可以由外层工作流管理重试、定时器和任务恢复。Workflow 的确定性部分读取事件历史，模型与网络等非确定性操作由 Activity 执行并记录结果。

Saga 补偿需要业务定义，例如创建资源失败后删除已创建部分。调度框架不会凭空推导撤销动作，也不自动提供跨所有系统的原子事务。重复 Activity 执行仍需幂等和副作用状态核对。

多 Agent

复杂任务可以把 Triage、Coder、Tester、Reviewer 分开，分别限制上下文、工具和输出。这样能减少单一 Agent 中过多规则与能力的混杂，也更容易独立评测。

但多个 Agent 可能共享同一种模型错误，辩论一致不等于结论正确。应根据任务复杂度、成本和可验证性决定是否拆分，不必把所有流程都改成 Multi-Agent Swarm。

拓扑 组织方式 主要权衡
Supervisor 中央节点分派并汇总 控制清楚，主管可能成为瓶颈
Hierarchical 多级主管分别管理领域 适合大任务，增加层级与传递成本
Peer-to-Peer Agent 直接 Handoff 交接灵活，需要全局环路和权限限制

中央调度不能自动杜绝死锁，对等拓扑也不必然失控。关键是明确谁拥有当前任务、怎样汇总、怎样结束以及怎样限制往返。

Handoff

Handoff把当前控制权交给另一 Agent，区别于“调用专家工具后由主管继续回答”。交接需要目标身份、任务、允许工具、上下文范围和返回约定。

Swarm 中返回另一 Agent 对象的工具函数可以说明这一抽象。下面用与框架无关的数据结构表达相同职责：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from dataclasses import dataclass


@dataclass(frozen=True)
class Handoff:
    """记录控制权转移目标以及允许携带的有限上下文。"""

    target_agent: str
    task_id: str
    context: dict


def transfer_to_reviewer(task_id: str, patch_id: str) -> Handoff:
    """只传递审查所需产物标识，不传入 Coder 的全部草稿。"""
    return Handoff(
        target_agent="reviewer",
        task_id=task_id,
        context={"patch_id": patch_id},
    )

Host 识别交接请求后检查目标是否注册、权限是否允许，以及交接次数是否超限，再切换活动 Agent。OpenAI Agents SDK 等实现提供相应编排与 Handoff 能力，具体 API 与教学原型需要区分。

状态树

主状态保存用户目标、任务 ID、Base Commit 和已核准产物；子状态保存局部检索、AST、草稿和内部循环。子 Agent 结束时，通过 Mapper 只把约定结果写回主图。

直接把所有 Agent 的长消息数组共享，会传播无关信息、过时方案和敏感数据。子状态隔离既减少上下文，也使组件测试具有明确输入输出。

编译后的 Subgraph 可以作为父图节点。状态一致时可直接组合；状态不同则需要输入、输出映射。子图可以继承或配置检查点，但私有状态是否保留、清理以及如何访问，仍由运行与存储策略决定。

协商与通信

Coder 提出补丁，Tester 返回运行证据，Reviewer 检查漏洞和架构约束。遇到 Critical 问题可以退回修复，超过讨论轮数则进入人工裁决。共识只能作为信号，测试和业务断言仍是独立依据。

跨语言或跨服务协作需要统一消息 Schema：

1
2
3
4
5
6
7
8
9
10
11
12
13
{
  "message_id": "message-42",
  "task_id": "repair-17",
  "trace_id": "trace-9",
  "sender_agent": "tester",
  "recipient_agent": "supervisor",
  "schema_version": 1,
  "payload": {
    "patch_id": "patch-3",
    "status": "failed",
    "failed_tests": ["test_empty_input"]
  }
}

RabbitMQ、NATS、Redis 等可用于消息协作，但能力不同：Redis Pub/Sub 不能直接视为具有持久重投能力的工作队列。需要定义确认、重试、去重、顺序与过期，避免同一消息重复触发操作。

LangGraph 适合显式图状态与恢复，AutoGen 常围绕 Agent 消息与协作组织，CrewAI 提供角色、任务和流程抽象。实际控制能力取决于版本及实现，不能用固定排名替代系统需求分析。

Sandbox

Sandbox限制代码能访问的资源，Harness组织任务、依赖、判分和回放。它们解决不同问题：隔离环境可以降低执行风险，但不证明算法正确；测试通过也不证明代码具有无限制执行权。

方案 架构特点 使用时核对
Docker 容器化环境，通常共享内核 用户、挂载、Capability、网络和配额
Firecracker MicroVM 虚拟化隔离 虚拟化支持、镜像与生命周期
E2B 面向代码执行的云沙箱服务 SDK、网络策略、超时与数据位置
Modal 托管计算与执行环境 权限、镜像、资源和清理配置
gVisor 用户态内核接口隔离 兼容性、开销及具体部署

不能把 gVisor 直接等同于 KVM MicroVM，也不能假设一个 with 语句在所有 SDK 版本中都自动销毁远端资源。执行器应明确创建、准备、运行、采集与清理，并记录清理失败。

执行边界

生成代码应进入独立工作目录或临时环境。源代码可只读挂载，输出使用专门可写目录，默认限制出站网络，环境变量只传必要项。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
services:
  sandbox:
    image: python:3.12-slim
    user: "65534:65534"
    read_only: true
    network_mode: none
    cap_drop:
      - ALL
    security_opt:
      - no-new-privileges:true
    mem_limit: 256m
    cpus: 1.0
    pids_limit: 64
    volumes:
      - ./candidate:/workspace:ro
    tmpfs:
      - /tmp:size=64m
    working_dir: /workspace
    environment:
      PYTHONDONTWRITEBYTECODE: "1"
    command: ["python", "-m", "pytest", "-p", "no:cacheprovider", "-q"]

这段配置表达隔离思路，并非所有项目都能不加调整直接运行。依赖需要预先准备，测试可能有特定写目录；限制网络不应通过继承宿主凭证来补救。Capability 与只读配置也不构成绝对安全证明。

超时要确认真实进程或容器已停止，输出需要流式预算，测试结束后恢复或销毁环境。临时快照应带项目版本，避免不同任务共享污染状态。

Harness 回放

确定性测试优先用 Mock 替换外部副作用，并录制模型、工具和环境中的非确定性响应。只固定 temperature 或 seed，不能保证跨模型服务和运行环境逐字重现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from collections import deque


class FixtureTool:
    """按顺序验证预期请求，并返回录制的工具观察。"""

    def __init__(self, records: list[dict]):
        self.records = deque(records)

    def call(self, name: str, arguments: dict) -> dict:
        """未匹配请求即失败，不偷偷回退到真实外部工具。"""
        if not self.records:
            raise AssertionError("Unexpected extra tool call")
        expected = self.records.popleft()
        if expected["name"] != name or expected["arguments"] != arguments:
            raise AssertionError("Tool call does not match the fixture")
        return expected["result"]

    def assert_complete(self) -> None:
        """确认所有必需调用都发生，防止漏执行却通过测试。"""
        if self.records:
            raise AssertionError("Expected calls were not completed")

Fixture 可记录输入 State、模型响应、Tool Call、Patch、Base Commit、环境镜像与测试报告。真实模型质量评估仍需另行调用模型，不能把纯回放的稳定性分数当成新模型生成能力。

评估指标

Agent 评估包含正确性、轨迹效率、成本和工具选择。若共采样 nn 次、正确 cc 次，选取 kk 个候选至少有一个成功的估计为：

Pass@k^=1−(n−ck)(nk).\widehat{\mathrm{Pass@k}} =1-\frac{\binom{n-c}{k}}{\binom{n}{k}}. 

1
2
3
4
5
6
7
8
9
10
from math import comb


def pass_at_k(n: int, c: int, k: int) -> float:
    """检查采样范围，再计算至少一个候选成功的组合数估计。"""
    if n <= 0 or not 0 <= c <= n or not 1 <= k <= n:
        raise ValueError("Require n>0, 0<=c<=n and 1<=k<=n")
    if n - c < k:
        return 1.0
    return 1.0 - comb(n - c, k) / comb(n, k)

Pass@1 对应 c/nc/n。这是采样成功率估计，不是对单个补丁的确定性证明。测试集、环境和判分规则需要固定，并报告样本量与失败类别。

轨迹效率记录成功任务的步骤、重试和 Handoff；成本区分输入、输出、缓存及工具资源；Tool Precision/Recall/F1 判断能力选择，不能只看最终回复。费用应使用实际模型单价和用量，不能给全部 Token 乘一个永久固定价格。

重构案例

一个完整流程可以由 Triage、Coder、Tester、Reviewer 四类 Agent 组成：先确定范围，Coder 生成受限 Patch，Tester 在独立环境运行测试，Reviewer 核对规则，再由主管选择提交、修复或停止。

连续失败后可以从历史分支选择其他策略，但剩余全局预算仍需减少，不能通过 Time Travel 重置计数继续无限尝试。回放控制层应在当前运行之外协调，避免节点递归调用整条图形成另一层失控循环。

Trace 记录节点、参数摘要、版本、耗时和结果，Fixture 支持诊断回放；CI 检查任务质量、安全、轨迹与成本。是否自动提交或合并，还需要仓库规则、真实授权和针对当前 Commit 的验证。

Graph 表达状态与路径，Loop Guard 限制无进展循环，Checkpoint 保存恢复位置，Sandbox 限制执行范围，Harness 验证任务行为。它们共同构成可测试、可恢复的 Agent 运行体系。