Agent Loop 恢复执行时如何避免重复副作用？

需要持久化 Invocation ID、Call ID 和业务幂等键。恢复通常是至少一次语义，外部写操作必须由拥有该数据的下游服务去重；无法确认结果时应进入 outcome_unknown 并先对账，而不是盲目重试。

Agent Loop 是什么？AI Agent 闭环运行机制详解

Q: Agent Loop 是什么？

Agent Loop 是 AI Agent 在一次任务执行中的有界运行循环：运行时把目标和已提交证据交给模型，接收类型化决策，校验并执行获准的工具调用，持久化结果，再继续执行或以明确原因终止。

Q: Agent Loop 和普通 while 循环有什么区别？

普通 while 循环按照程序员写死的确定性条件执行；Agent Loop 由模型提出下一步语义动作，但权限、预算、状态迁移、副作用策略和终止条件仍由可信代码强制执行。

Q: Agent Loop 应该如何终止？

运行时应返回明确的类型化原因，例如 completed、approval_required、cancelled、deadline_exceeded、max_steps、tool_budget_exhausted、no_progress、invalid_decision、model_error 或 checkpoint_failed。最终回答只是其中一种终止路径。

Q: Agent Loop 可以并行调用工具吗？

可以，但调用必须相互独立且符合并发策略。每个请求都要有唯一 Call ID，结果必须引用同一个 ID，运行时应收齐并提交完整结果批次后再让模型进入下一轮。

2026-07-04 - QubitTool 技术团队

核心摘要

Agent Loop 是 AI Agent 执行多步任务时使用的有界运行协议。 每一轮由可信代码装载目标、状态和观察，模型只提出类型化的回答、工具批次或升级请求；运行时负责校验、执行、关联结果并持久化事件，最后继续下一轮或返回明确的终止原因。生产级 Agent Loop 的核心不是在 LLM 外套一层无限 while true，而是把决策权与控制权分开。

核心要点

模型负责提议，Runtime 负责裁决：模型输出在通过类型、Schema、权限、预算和副作用策略校验前，都只是不可信提议。
一轮循环是一笔协议事务：决策、校验、执行、关联、提交和恢复是不同阶段，失败语义不能混在一起。
终止原因必须类型化：completed、cancelled、approval_required 和 no_progress 代表不同状态，不能统一压成 success: false。
并行调用必须可关联：Call ID 要从请求贯穿到结果，收齐完整结果批次并提交后，模型才能进入下一轮。
Checkpoint 不等于副作用账本：对话状态已经持久化，不能证明邮件、付款、删除等外部写操作只执行了一次。
评测对象是完整轨迹：工具选择、副作用安全、进展、成本、延迟和停止原因都属于任务质量，不能只看最终文字。

Agent Loop 定义

Agent Loop 是 AI Agent 在一次任务调用中反复执行的控制循环：

text

目标与已提交状态
→ 模型决策
→ 运行时校验
→ 工具执行
→ 按 Call ID 关联观察
→ 持久化提交
→ 继续或终止

这个定义刻意收窄了边界。队列调度、Worker 租约、长期记忆、部署和实例管理属于更广义的 Agent Runtime；跨多次运行做 Prompt、工具与评测迭代属于 Loop Engineering。本文只解释“一次任务执行内部”的控制协议。

Agent Loop 和普通 while 循环

普通 while 循环由代码判断确定性条件。Agent Loop 允许概率模型提出下一步语义动作，但循环边界必须继续由确定性代码掌握。

责任	普通 while 循环	生产级 Agent Loop
下一步动作	程序逻辑决定	模型提出
允许的状态迁移	程序逻辑决定	Runtime 状态机决定
输入合法性	通常由本地类型保证	必须执行 Schema 校验
权限	被调用代码决定	Runtime 与下游服务共同决定
停止条件	布尔表达式	类型化终止策略
崩溃恢复	业务自行处理	Checkpoint 加副作用对账
正确性证据	返回值与测试	结果、轨迹与真实副作用

因此，正确的心智模型不是“LLM 控制循环”，而是“LLM 是可信循环中的一个决策组件”。

Agent Loop 执行协议

稳健的一轮 Agent Loop 包含八个阶段，而且执行顺序本身就是正确性要求。

阶段	Runtime 的责任	应持久化的证据
1. 装载	读取目标、策略版本、Checkpoint 和剩余预算	Invocation ID、State Version
2. 构造上下文	选择相关历史、工具 Schema 和最新观察	上下文引用或摘要 Hash
3. 决策	请求模型返回一个类型化决策	模型响应 ID、决策类型
4. 校验	检查 Schema、Call ID、工具、授权和预算	校验结果、策略原因
5. 执行	在取消与超时控制下运行获准的调用批次	Attempt ID、Effect ID
6. 关联	将每个结果绑定到原始 Call ID	完整结果批次
7. 提交	原子追加本轮事件并推进状态版本	新的持久化 Checkpoint
8. 继续或停止	基于已提交状态进入下一轮，或返回终止原因	Terminal Reason、结果摘要

提交边界不能省略。如果工具结果还没有持久化，Runtime 就再次询问模型，崩溃恢复后可能得到一段不同历史。Google ADK 的运行时文档同样把状态变更放入 Event 提交后，再继续后续执行。

使用封闭的决策类型

不要让模型返回任意自然语言，再靠关键词猜测是否要调用工具。内部协议应该使用封闭联合类型：

text

final(answer)
tool_calls([{call_id, tool_name, arguments}])
request_approval(proposal_digest)
clarification(question)
cannot_continue(reason)

不同模型供应商会使用不同消息块和停止字段。Adapter 应把它们转换成统一的内部契约，让供应商格式停留在边缘，避免渗透到核心状态机。

区分协议状态与业务状态

协议状态回答“执行走到了哪里”，业务状态回答“任务已经知道或改变了什么”。

协议状态	业务状态
Invocation ID	用户请求
Turn、State Version	已提取事实
Pending Call ID	领域对象
剩余预算	校验结论
Terminal Reason	获准结果
Worker Lease	外部副作用状态

两者混在一起会让恢复变得危险。Checkpoint 中的一句“退款已完成”，不是支付服务已经提交退款的权威证据。

ReAct 与其他 Agent 模式

ReAct、Plan-and-Execute 和 Reflexion 都可以运行在 Agent Loop 内部，但都不能替代运行协议。

模式	它改变什么	Runtime 仍必须负责什么
ReAct	让推理与行动交替	校验、工具、状态、预算、终止
Plan-and-Execute	先规划，再逐步执行	计划版本、权限、重规划上限
Reflexion	利用反馈修正后续行为	反馈来源、重试边界、验收测试
状态机	把决策限制在显式状态中	工具执行、持久化提交、副作用安全
图工作流	在预定义节点间路由	节点内循环、共享状态、取消传播

原始 ReAct 论文说明了交错推理、行动与观察的价值，但它描述的是认知模式，不是生产授权模型。可靠运行也不要求保存模型的隐藏思维链；应该记录类型化决策、证据引用、工具结果、策略判定和终止原因。

Anthropic 在 Building Effective Agents 中区分了代码预定义的 Workflow 和由模型动态决定流程的 Agent，并建议从能解决问题的最简单方案开始。只要模型可以选择下一步动作，本文的 Runtime 控制对两类方案都适用。

生产状态机

生产级 Agent Loop 应把每条继续路径和退出路径都画成显式状态。

flowchart TD A["装载已提交状态"] --> B{"已取消或预算耗尽"} B -->|是| Z["返回类型化终止原因"] B -->|否| C["构造有界上下文"] C --> D["模型提出决策"] D --> E{"决策类型"} E -->|最终回答| F["校验最终回答"] E -->|工具批次| G["校验 ID、Schema、策略和预算"] E -->|需要审批| H["持久化提议摘要"] E -->|非法决策| Z G --> I{"调用相互独立"} I -->|是| J["并发执行"] I -->|否| K["按声明顺序执行"] J --> L["按 Call ID 关联结果"] K --> L L --> M["提交完整 Turn Event"] M --> N{"检测到进展"} N -->|是| B N -->|否| Z F --> O["提交最终事件"] O --> Z H --> Z

状态机不能只暴露一个笼统的 success: false。调用方和运维人员需要稳定的终止词表。

终止原因	含义	调用方通常如何处理
`completed`	已提交通过校验的最终回答	返回结果
`approval_required`	合法动作触发策略审批门	展示精确的 Proposal
`cancelled`	调用方或租约取消执行	禁止再启动新副作用
`deadline_exceeded`	整体墙钟时间到期	只在调用方策略允许时重试
`max_steps`	Turn 预算耗尽	检查轨迹或收窄目标
`tool_budget_exhausted`	工具调用数量达到上限	有数据支持时再增加
`no_progress`	等价动作批次重复出现	升级、重规划或停止
`invalid_decision`	模型输出违反内部协议	有界修复一次或明确失败
`model_error`	模型请求失败	使用供应商级重试策略
`checkpoint_failed`	无法提交持久化状态	恢复前先核对已发起动作

可运行的 Go 控制器

下面的标准库 Go 程序实现控制面，不绑定具体模型 SDK。它会校验类型化决策、保留 Call ID、并发执行独立工具、在下一轮模型调用前提交观察、强制执行预算、检测重复计划、响应取消，并返回类型化终止原因。

为了聚焦核心机制，示例中的模型 Adapter 只接受 final 与 tool_calls，审批由 Runtime Policy 强制执行。完整 Adapter 可以增加澄清和模型主动拒绝等类型，而不改变提交与执行不变量。

package main

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"
)

type DecisionKind string

const (
	DecisionFinal DecisionKind = "final"
	DecisionTools DecisionKind = "tool_calls"
)

type TerminalReason string

const (
	Completed           TerminalReason = "completed"
	ApprovalRequired    TerminalReason = "approval_required"
	Cancelled           TerminalReason = "cancelled"
	DeadlineExceeded    TerminalReason = "deadline_exceeded"
	MaxSteps            TerminalReason = "max_steps"
	ToolBudgetExhausted TerminalReason = "tool_budget_exhausted"
	NoProgress          TerminalReason = "no_progress"
	InvalidDecision     TerminalReason = "invalid_decision"
	ModelError          TerminalReason = "model_error"
	CheckpointFailed    TerminalReason = "checkpoint_failed"
)

var ErrApprovalRequired = errors.New("approval required")

type ToolCall struct {
	ID        string          `json:"id"`
	Name      string          `json:"name"`
	Arguments json.RawMessage `json:"arguments"`
}

type Decision struct {
	Kind   DecisionKind `json:"kind"`
	Answer string       `json:"answer,omitempty"`
	Calls  []ToolCall   `json:"calls,omitempty"`
}

type ToolResult struct {
	CallID string          `json:"call_id"`
	Name   string          `json:"name"`
	Output json.RawMessage `json:"output,omitempty"`
	Error  string          `json:"error,omitempty"`
}

type Event struct {
	Step         int          `json:"step"`
	StateVersion int          `json:"state_version"`
	Decision     Decision     `json:"decision"`
	Results      []ToolResult `json:"results,omitempty"`
}

type State struct {
	Goal         string
	Step         int
	StateVersion int
	ToolCalls    int
	Events       []Event
	LastPlan     string
	RepeatStreak int
}

type Budget struct {
	MaxSteps        int
	MaxToolCalls    int
	MaxRepeatedPlan int
	Deadline        time.Time
}

type Outcome struct {
	Reason TerminalReason
	Answer string
	State  State
	Err    error
}

type Tool interface {
	Execute(context.Context, json.RawMessage) (json.RawMessage, error)
}

type Controller struct {
	Budget    Budget
	Decide    func(context.Context, State) (Decision, error)
	Tools     map[string]Tool
	Authorize func(State, []ToolCall) error
	Append    func(context.Context, Event) error
}

func contextOutcome(state State, err error) Outcome {
	reason := Cancelled
	if errors.Is(err, context.DeadlineExceeded) {
		reason = DeadlineExceeded
	}
	return Outcome{Reason: reason, State: state, Err: err}
}

func (c Controller) Run(ctx context.Context, state State) Outcome {
	runCtx := ctx
	var cancel context.CancelFunc = func() {}
	if !c.Budget.Deadline.IsZero() {
		runCtx, cancel = context.WithDeadline(ctx, c.Budget.Deadline)
	}
	defer cancel()

	for {
		if err := runCtx.Err(); err != nil {
			return contextOutcome(state, err)
		}
		if state.Step >= c.Budget.MaxSteps {
			return Outcome{Reason: MaxSteps, State: state}
		}

		decision, err := c.Decide(runCtx, state)
		if err != nil {
			if ctxErr := runCtx.Err(); ctxErr != nil {
				return contextOutcome(state, ctxErr)
			}
			return Outcome{Reason: ModelError, State: state, Err: err}
		}

		switch decision.Kind {
		case DecisionFinal:
			if strings.TrimSpace(decision.Answer) == "" || len(decision.Calls) != 0 {
				return Outcome{Reason: InvalidDecision, State: state}
			}
			if err := c.commit(runCtx, &state, decision, nil); err != nil {
				if ctxErr := runCtx.Err(); ctxErr != nil {
					return contextOutcome(state, ctxErr)
				}
				return Outcome{Reason: CheckpointFailed, State: state, Err: err}
			}
			return Outcome{Reason: Completed, Answer: decision.Answer, State: state}

		case DecisionTools:
			if err := c.validateCalls(decision.Calls); err != nil {
				return Outcome{Reason: InvalidDecision, State: state, Err: err}
			}
			if state.ToolCalls+len(decision.Calls) > c.Budget.MaxToolCalls {
				return Outcome{Reason: ToolBudgetExhausted, State: state}
			}
			if c.Authorize != nil {
				if err := c.Authorize(state, decision.Calls); err != nil {
					if errors.Is(err, ErrApprovalRequired) {
						return Outcome{Reason: ApprovalRequired, State: state, Err: err}
					}
					return Outcome{Reason: InvalidDecision, State: state, Err: err}
				}
			}

			fingerprint, err := batchFingerprint(decision.Calls)
			if err != nil {
				return Outcome{Reason: InvalidDecision, State: state, Err: err}
			}
			if fingerprint == state.LastPlan {
				state.RepeatStreak++
			} else {
				state.LastPlan = fingerprint
				state.RepeatStreak = 1
			}
			if state.RepeatStreak > c.Budget.MaxRepeatedPlan {
				return Outcome{Reason: NoProgress, State: state}
			}

			results := c.executeParallel(runCtx, decision.Calls)
			if err := c.commit(runCtx, &state, decision, results); err != nil {
				if ctxErr := runCtx.Err(); ctxErr != nil {
					return contextOutcome(state, ctxErr)
				}
				return Outcome{Reason: CheckpointFailed, State: state, Err: err}
			}
			state.ToolCalls += len(decision.Calls)

		default:
			return Outcome{Reason: InvalidDecision, State: state}
		}
	}
}

func (c Controller) validateCalls(calls []ToolCall) error {
	if len(calls) == 0 {
		return errors.New("empty tool batch")
	}
	seen := make(map[string]struct{}, len(calls))
	for _, call := range calls {
		if call.ID == "" || call.Name == "" || !json.Valid(call.Arguments) {
			return fmt.Errorf("invalid tool call %q", call.ID)
		}
		if _, ok := seen[call.ID]; ok {
			return fmt.Errorf("duplicate call ID %q", call.ID)
		}
		if _, ok := c.Tools[call.Name]; !ok {
			return fmt.Errorf("unknown tool %q", call.Name)
		}
		seen[call.ID] = struct{}{}
	}
	return nil
}

func (c Controller) executeParallel(ctx context.Context, calls []ToolCall) []ToolResult {
	type indexedResult struct {
		index  int
		result ToolResult
	}

	ch := make(chan indexedResult, len(calls))
	var wg sync.WaitGroup
	for index, call := range calls {
		wg.Add(1)
		go func(index int, call ToolCall) {
			defer wg.Done()
			output, err := c.Tools[call.Name].Execute(ctx, call.Arguments)
			result := ToolResult{CallID: call.ID, Name: call.Name, Output: output}
			if err != nil {
				result.Output = nil
				result.Error = err.Error()
			}
			ch <- indexedResult{index: index, result: result}
		}(index, call)
	}
	wg.Wait()
	close(ch)

	indexed := make([]indexedResult, 0, len(calls))
	for result := range ch {
		indexed = append(indexed, result)
	}
	sort.Slice(indexed, func(i, j int) bool {
		return indexed[i].index < indexed[j].index
	})

	results := make([]ToolResult, 0, len(indexed))
	for _, result := range indexed {
		results = append(results, result.result)
	}
	return results
}

func (c Controller) commit(
	ctx context.Context,
	state *State,
	decision Decision,
	results []ToolResult,
) error {
	event := Event{
		Step:         state.Step + 1,
		StateVersion: state.StateVersion + 1,
		Decision:     decision,
		Results:      results,
	}
	if err := c.Append(ctx, event); err != nil {
		return err
	}
	state.Step = event.Step
	state.StateVersion = event.StateVersion
	state.Events = append(state.Events, event)
	return nil
}

func batchFingerprint(calls []ToolCall) (string, error) {
	type normalizedCall struct {
		Name      string          `json:"name"`
		Arguments json.RawMessage `json:"arguments"`
	}

	normalized := make([]normalizedCall, 0, len(calls))
	for _, call := range calls {
		var compacted bytes.Buffer
		if err := json.Compact(&compacted, call.Arguments); err != nil {
			return "", err
		}
		normalized = append(normalized, normalizedCall{
			Name:      call.Name,
			Arguments: append(json.RawMessage(nil), compacted.Bytes()...),
		})
	}
	sort.Slice(normalized, func(i, j int) bool {
		if normalized[i].Name == normalized[j].Name {
			return string(normalized[i].Arguments) < string(normalized[j].Arguments)
		}
		return normalized[i].Name < normalized[j].Name
	})
	payload, err := json.Marshal(normalized)
	if err != nil {
		return "", err
	}
	sum := sha256.Sum256(payload)
	return hex.EncodeToString(sum[:]), nil
}

type lookupTool struct {
	values map[string]string
}

func (t lookupTool) Execute(
	ctx context.Context,
	arguments json.RawMessage,
) (json.RawMessage, error) {
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	default:
	}

	var input struct {
		Key string `json:"key"`
	}
	if err := json.Unmarshal(arguments, &input); err != nil {
		return nil, err
	}
	value, ok := t.values[input.Key]
	if !ok {
		return nil, fmt.Errorf("key %q not found", input.Key)
	}
	return json.Marshal(struct {
		Key   string `json:"key"`
		Value string `json:"value"`
	}{Key: input.Key, Value: value})
}

func scriptedDecision(_ context.Context, state State) (Decision, error) {
	if len(state.Events) == 0 {
		return Decision{
			Kind: DecisionTools,
			Calls: []ToolCall{
				{ID: "call-profile", Name: "lookup", Arguments: json.RawMessage(`{"key":"profile"}`)},
				{ID: "call-policy", Name: "lookup", Arguments: json.RawMessage(`{"key":"policy"}`)},
			},
		}, nil
	}

	values := make(map[string]string)
	for _, result := range state.Events[len(state.Events)-1].Results {
		var output struct {
			Key   string `json:"key"`
			Value string `json:"value"`
		}
		if result.Error != "" {
			return Decision{}, errors.New(result.Error)
		}
		if err := json.Unmarshal(result.Output, &output); err != nil {
			return Decision{}, err
		}
		values[output.Key] = output.Value
	}
	return Decision{
		Kind:   DecisionFinal,
		Answer: fmt.Sprintf("profile=%s; policy=%s", values["profile"], values["policy"]),
	}, nil
}

func main() {
	events := make([]Event, 0, 2)
	controller := Controller{
		Budget: Budget{
			MaxSteps:        4,
			MaxToolCalls:    4,
			MaxRepeatedPlan: 1,
		},
		Decide: scriptedDecision,
		Tools: map[string]Tool{
			"lookup": lookupTool{values: map[string]string{
				"profile": "active",
				"policy":  "refund-under-100",
			}},
		},
		Append: func(_ context.Context, event Event) error {
			events = append(events, event)
			return nil
		},
	}

	outcome := controller.Run(context.Background(), State{
		Goal: "Load the customer profile and refund policy",
	})
	fmt.Println(outcome.Reason)
	fmt.Println(outcome.Answer)
	fmt.Printf(
		"steps=%d tool_calls=%d events=%d\n",
		outcome.State.Step,
		outcome.State.ToolCalls,
		len(events),
	)
}

预期输出：

text

completed
profile=active; policy=refund-under-100
steps=2 tool_calls=2 events=2

示例中的两个工具都是只读操作。生产 Adapter 还应校验各工具的参数 Schema、设置单次调用超时、对持久化 Payload 做脱敏，并在允许并发前识别有副作用的工具。

确定性的控制面测试

真实模型输出可能变化，但 Controller 的不变量不能变化。测试时注入脚本化决策和 Fake Tool，覆盖结果关联、预算、取消、审批、重复计划与 Checkpoint 失败。

package main

import (
	"context"
	"encoding/json"
	"sync/atomic"
	"testing"
	"time"
)

type delayedLookup struct {
	values map[string]string
}

func (t delayedLookup) Execute(
	ctx context.Context,
	arguments json.RawMessage,
) (json.RawMessage, error) {
	var input struct {
		Key string `json:"key"`
	}
	if err := json.Unmarshal(arguments, &input); err != nil {
		return nil, err
	}
	if input.Key == "profile" {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(10 * time.Millisecond):
		}
	}
	return json.Marshal(struct {
		Key   string `json:"key"`
		Value string `json:"value"`
	}{Key: input.Key, Value: t.values[input.Key]})
}

func newTestController() Controller {
	return Controller{
		Budget: Budget{
			MaxSteps:        4,
			MaxToolCalls:    4,
			MaxRepeatedPlan: 1,
		},
		Decide: scriptedDecision,
		Tools: map[string]Tool{
			"lookup": delayedLookup{values: map[string]string{
				"profile": "active",
				"policy":  "refund-under-100",
			}},
		},
		Append: func(context.Context, Event) error {
			return nil
		},
	}
}

type countingTool struct {
	executions *atomic.Int32
}

func (t countingTool) Execute(
	_ context.Context,
	_ json.RawMessage,
) (json.RawMessage, error) {
	t.executions.Add(1)
	return json.RawMessage(`{"ok":true}`), nil
}

func newRepeatingTestController(executions *atomic.Int32) Controller {
	return Controller{
		Budget: Budget{
			MaxSteps:        4,
			MaxToolCalls:    4,
			MaxRepeatedPlan: 1,
		},
		Decide: func(context.Context, State) (Decision, error) {
			return Decision{
				Kind: DecisionTools,
				Calls: []ToolCall{{
					ID:        "same-call",
					Name:      "count",
					Arguments: json.RawMessage(`{"key":"same"}`),
				}},
			}, nil
		},
		Tools: map[string]Tool{
			"count": countingTool{executions: executions},
		},
		Append: func(context.Context, Event) error {
			return nil
		},
	}
}

func TestParallelResultsPreserveCallOrder(t *testing.T) {
	controller := newTestController()
	outcome := controller.Run(context.Background(), State{Goal: "test"})

	if outcome.Reason != Completed {
		t.Fatalf("reason = %s, want %s", outcome.Reason, Completed)
	}
	results := outcome.State.Events[0].Results
	if results[0].CallID != "call-profile" || results[1].CallID != "call-policy" {
		t.Fatalf("unexpected result order: %#v", results)
	}
}

func TestRepeatedPlanStopsBeforeSecondExecution(t *testing.T) {
	var executions atomic.Int32
	controller := newRepeatingTestController(&executions)
	outcome := controller.Run(context.Background(), State{Goal: "test"})

	if outcome.Reason != NoProgress {
		t.Fatalf("reason = %s, want %s", outcome.Reason, NoProgress)
	}
	if executions.Load() != 1 {
		t.Fatalf("executions = %d, want 1", executions.Load())
	}
}

这种测试不要求模型逐字复现某段思维过程，而是验证模型之外必须保持确定性的边界。

并行工具调用

只有策略确认调用相互独立时，Agent Loop 才能安全并发执行，不能仅凭模型一次返回了多个 Tool Call 就并行。

OpenAI 的 Agents SDK 运行循环文档描述了单轮模型可能产生多个 Tool Call。Anthropic 的 Tool Use 文档为每个 tool_use 分配唯一 ID，并要求对应 tool_result 引用该 ID。两种线协议不同，但可以抽取出相同的 Runtime 不变量：

每个调用都有唯一且稳定的 Call ID。
结果必须携带原 Call ID。
不能只靠数组下标关联请求与结果。
下一轮模型调用应看到完整结果批次。
是否并发由 Runtime 判断，不是模型自行获得的权限。

哪些调用可以并发

当调用只读或满足交换律时，才适合并发：任一调用都不会改变另一个调用读取或写入的数据，顺序不影响结果，并且两者可以独立重试。

调用批次	是否并发	原因
读取客户资料 + 读取退款策略	通常可以	两个独立读操作
检索两个无关索引	通常可以	没有共享写入
创建订单 + 对该订单扣款	不可以	扣款依赖新订单 ID
更新余额 + 发送最终回执	不可以	回执依赖已提交余额
同时写同一份文档	不可以	顺序和冲突策略会改变结果

面对混合批次，应该构建依赖图，或者拒绝本次批次并要求 Planner 显式排序，不能靠工具名称猜依赖关系。

不要抹平局部失败

同一批次可能出现两个成功和一个失败。每个结果都应独立保留：

json

[
  {"call_id":"call-a","name":"read_profile","output":{"tier":"pro"}},
  {"call_id":"call-b","name":"read_policy","error":"deadline exceeded"},
  {"call_id":"call-c","name":"read_balance","output":{"amount":42}}
]

下一轮可以只重试 call-b，也可以利用已成功的证据或直接终止。如果压缩成一个批次级 failed，模型会丢失这些选择，并更容易制造重复调用。

终止条件与进展判断

Agent Loop 的结束原因可分为协议、策略与基础设施三类，只有第一类中的最终回答属于正常回答路径。

协议终止

模型返回 final，且答案通过输出校验。
模型返回带结构化原因的 cannot_continue。
当前歧义必须由用户澄清。

策略终止

工具调用需要人工审批。
调用超出 Tenant、对象、金额或 Capability 权限。
Step、Token、Tool Call、成本或墙钟时间预算耗尽。
相同或等价计划持续重复，没有产生新证据。

基础设施终止

模型供应商在重试预算内仍然失败。
必要工具持续不可用。
Checkpoint 无法持久化。
Worker 丢失租约或收到取消信号。

max_steps 只是最后一道安全网，不能代替进展定义。连续执行八个不同但无用的调用，仍然是卡住。

检测语义上的无进展

实用的进展检测器会组合多种信号：

信号	示例
重复动作指纹	同一 Tool 与归一化参数再次出现
状态增量	没有新增事实、产物或校验结果
目标谓词	必须满足的验收条件始终不变
错误循环	相同修复动作后仍出现同类错误
证据新颖度	新结果与已提交证据重复

第一版应从完全确定性的精确指纹开始。语义相似度只能作为二级信号，而且需要用 Replay 数据验证误停率；轮询和分页等合法任务本来就可能出现相似动作。

崩溃恢复与副作用

崩溃恢复通常是至少一次语义，业务副作用的“恰好一次”只能由业务服务边界实现。

Google ADK 的 Resume 文档明确讨论了至少一次恢复和避免重复 Tool 执行的问题。这个结论不依赖具体框架：进程崩溃后，Runtime 可能只知道请求已经发出，却不知道远端系统是否成功提交。

使用三类持久化身份

身份	作用域	用途
Invocation ID	整个任务	聚合所有 Turn 与重试
Call ID	一次逻辑工具请求	关联提议、Attempt 和结果
Effect Key	一次业务变更	在外部服务中去重副作用

Effect Key 必须跨网络重试和进程重启保持稳定，并由拥有该变更的下游服务强制执行。Agent 进程内的一张 Map 无法提供这个保证。

显式建模结果未知

假设付款请求到达服务端后，客户端等待超时。此时不能直接记为 failed：

text

prepared → dispatched → committed
                     ↘ outcome_unknown → reconcile → committed 或 rejected

从 outcome_unknown 直接重试可能导致重复扣款。应先用 Effect Key 向业务服务查询；如果服务不支持对账，就停止并转人工处理，不能伪造成功或失败结论。

Checkpoint 和副作用账本解决不同问题

Checkpoint：Agent Loop 已经持久化了哪些决策与观察。
副作用账本：业务服务接受、拒绝或暂时无法确认了哪些外部变更。

发送请求前持久化 Call Proposal，完成后持久化关联结果，并且只从已提交状态恢复。高风险写操作应使用 Transactional Outbox 或由下游维护幂等记录，避免状态提交与网络派发静默分叉。

取消也遵循同一条所有权原则：取消 Invocation，撤销或过期 Worker Lease，并要求工具在启动下一次副作用前检查 context.Context。已经提交的业务副作用不能被取消信号自动撤销。

如何评测 Agent Loop

Agent Loop 评测必须同时覆盖任务、轨迹、副作用和运行指标，只评价最终回答会奖励不安全或浪费资源的执行路径。

维度	指标	回答的问题
任务	分场景成功率	用户要求的结果是否真正发生
决策	Tool 与参数正确率	模型是否提出正确动作
效率	成功任务的 Turn、Call、Token 与成本	完成结果用了多少资源
进展	No-progress 与重复计划率	循环是否持续接近目标
可靠性	工具错误恢复率、Resume 成功率	可控故障能否正确恢复
副作用	未授权或重复副作用率	是否只改变了允许改变的数据
终止	Stop Reason 准确率	是否在正确时间以正确原因停止
运行	按 Terminal Reason 拆分的 p50、p95、p99 延迟	时间消耗在哪个阶段

所有指标都应按任务类型和终止原因分布展示。一个快速的 approval_required、一个耗时较长但成功的研究任务，以及一个重复计划事故，不应该被全局平均值混成同一种表现。

从状态迁移构造评测集

评测集至少覆盖这些分支：

不调用工具直接回答
一次合法工具调用
独立并行调用乱序返回
参数格式错误和未知工具
批次局部失败
需要审批的写操作
等价计划重复
工具执行期间取消
派发后、结果提交前崩溃
携带已有 Effect Key 恢复
最终回答未通过确定性 Validator

更完整的 Trace、隐私、OpenTelemetry 边界和发布指标可继续阅读 Agent 可观测性工程。那篇文章负责观测平面，本文只定义循环本身应该产生什么可靠事件。

常见失败模式

把循环边界交给模型

现象：Prompt 写着“完成后停止”，但代码没有 Step、时间或 Tool Budget。

修复：在可信代码中执行预算与终止策略。Prompt 指令只是提示，不是资源控制。

并行结果没有 Call ID

现象：系统按数组顺序绑定结果，较慢的响应被当作另一个工具的输出。

修复：强制唯一 Call ID；缺失、重复或未知 ID 必须在提交前拒绝。

把所有错误都变成普通文本

现象：模型无法区分参数校验失败、超时、拒绝、结果未知和永久业务失败。

修复：保留结构化错误类别、Retryable、Attempt ID 和 Effect Status。

任何失败都自动重试

现象：非法参数、权限拒绝和结果未知的写操作，都按瞬时读失败处理。

修复：按操作类型和错误类别定义重试策略，不能允许模型把“拒绝”解释成“获得权限”。

用 Checkpoint 证明外部副作用

现象：状态写着邮件或退款已完成，但没有业务服务出具的权威 Receipt。

修复：保存业务服务返回的 Effect ID，并向拥有数据的服务对账。

只评价最终文字

现象：回答看起来合理就通过，即使 Loop 读取了错误客户、重复执行写操作或超出预算。

修复：把状态迁移、证据、副作用、预算和终止原因与最终回答一起评分。

生产检查清单

上线一个 Agent Loop 前，至少确认：

[ ] 模型返回封闭且带版本的决策 Schema。
[ ] 未知决策类型、未知工具、非法字段和重复 Call ID 都会被拒绝。
[ ] Tool 权限同时校验 Actor、Tenant、对象和操作。
[ ] 只有经过策略确认的独立调用才会并发执行。
[ ] 结果保留 Call ID 与局部失败。
[ ] 下一次模型调用前，完整 Turn Event 已持久化提交。
[ ] Step、Tool、Token、成本和墙钟时间预算都由代码执行。
[ ] 取消后 Worker 无法开始新的副作用。
[ ] 精确重复计划和目标谓词不变会触发 No-progress 处理。
[ ] 外部写操作使用稳定幂等键和业务服务 Receipt。
[ ] 未知结果进入 Reconciliation，不能盲目重试。
[ ] 类型化 Terminal Reason 对调用方与运维均可见。
[ ] Replay 测试覆盖所有关键状态迁移和失败分支。
[ ] Trace 默认记录有界证据，而不是隐藏思维链。

常见问题

Agent Loop 是什么

Agent Loop 是把模型从一次性回答器变成多步任务执行器的有界运行循环。每一轮先读取已提交状态，再让模型提出类型化决策；可信 Runtime 校验决策、执行获准工具、持久化关联观察，最后继续循环或返回明确终止原因。

Agent Loop 和普通 while 循环有什么区别

普通 while 循环的条件与动作都由确定性代码决定。Agent Loop 中，模型可以提出下一步语义动作，但状态机、Schema 校验、权限、预算、副作用策略、提交顺序、取消和终止仍归可信代码所有。在 LLM 外包一层没有这些控制的 while true，不能称为生产级 Agent Loop。

Agent Loop 应该如何终止

不要只返回一个 Success 布尔值。至少区分 completed、approval_required、cancelled、deadline_exceeded、max_steps、tool_budget_exhausted、no_progress、invalid_decision、model_error 和 checkpoint_failed，调用方才能正确选择返回、审批、重试、对账或升级处理。

Agent Loop 可以并行调用工具吗

可以，前提是调用相互独立且 Runtime 策略允许。每个调用都要分配唯一 ID，在取消和超时控制下执行，保留局部失败，恢复确定性顺序，并在下一轮模型调用前提交完整关联批次。存在依赖或写冲突的调用必须串行执行。

Agent Loop 恢复时如何避免重复副作用

持久化 Invocation、Call、Attempt 和 Effect 四类身份。恢复时从最后一个已提交 Event 开始，对已经派发但没有确定结果的 Effect Key 向业务服务查询。超时不等于失败，对话 Checkpoint 也不能证明业务变更已发生。至少一次编排必须搭配下游幂等或显式对账。

总结

Agent Loop 本质上是受协议控制的状态机，不是“自主运行”的无限循环。模型提出类型化决策；可信代码负责校验权限与预算、执行并关联工具、提交状态、检测进展，并返回精确的终止原因。只有明确依赖关系，并行才能真正降低延迟；只有把 Checkpoint 与业务服务维护的 Effect Identity 结合，崩溃恢复才不会制造重复副作用。

理解跨多次运行改进 Prompt、工具和策略的外循环，可继续阅读 Agent Loop 与 Loop Engineering 的区别；执行级调度、租约与持久化生命周期则由 Agent Runtime 术语承接。

一手来源

上一篇:Loop Engineering 实践指南：设计可靠的 Agent 自动化闭环

下一篇:Agent Loop 与 Loop Engineering：内外循环

Agent Loop 是什么？AI Agent 闭环运行机制详解

核心摘要

目录

核心要点

Agent Loop 定义

Agent Loop 和普通 while 循环

Agent Loop 执行协议

使用封闭的决策类型

区分协议状态与业务状态

ReAct 与其他 Agent 模式

生产状态机

可运行的 Go 控制器

确定性的控制面测试

并行工具调用

哪些调用可以并发

不要抹平局部失败

终止条件与进展判断

协议终止

策略终止

基础设施终止

检测语义上的无进展

崩溃恢复与副作用

使用三类持久化身份

显式建模结果未知

Checkpoint 和副作用账本解决不同问题

如何评测 Agent Loop

从状态迁移构造评测集

常见失败模式

把循环边界交给模型

并行结果没有 Call ID

把所有错误都变成普通文本

任何失败都自动重试

用 Checkpoint 证明外部副作用

只评价最终文字

生产检查清单

常见问题

Agent Loop 是什么

Agent Loop 和普通 while 循环有什么区别

Agent Loop 应该如何终止

Agent Loop 可以并行调用工具吗

Agent Loop 恢复时如何避免重复副作用

总结

一手来源